Fix handling null messages during timeout evaluation.

eureka-client/build.gradle

@@ -39,6 +39,6 @@ dependencies {
     testCompile "org.mock-server:mockserver-netty:${mockserverVersion}"
     testCompile "com.netflix.governator:governator:${governatorVersion}"
     testCompile "com.github.tomakehurst:wiremock-jre8:2.25.1"
-    testCompile "org.assertj:assertj-core:3.11.1"
+    testCompile "org.assertj:assertj-core:3.24.2"
     testCompile "javax.servlet:javax.servlet-api:4.0.1"
 }

eureka-core/build.gradle

@@ -21,4 +21,5 @@ dependencies {
     testCompile "com.jcraft:jzlib:1.1.3" // netty dependency
     testCompile "org.mockito:mockito-core:${mockitoVersion}"
     testRuntime 'org.slf4j:slf4j-simple:1.7.10'
+    testCompile "org.assertj:assertj-core:3.24.2"
 }

eureka-core/src/main/java/com/netflix/eureka/cluster/ReplicationTaskProcessor.java

@@ -192,8 +192,8 @@ private static boolean isNetworkConnectException(Throwable e) {
      */
     private static boolean maybeReadTimeOut(Throwable e) {
         do {
-            if (IOException.class.isInstance(e)) {
-            	String message = e.getMessage().toLowerCase();
+            if (e instanceof IOException) {
+                String message = e.getMessage() != null ? e.getMessage().toLowerCase() :  "";
             	Matcher matcher = READ_TIME_OUT_PATTERN.matcher(message);
             	if(matcher.find()) {
             		return true;

eureka-core/src/test/java/com/netflix/eureka/cluster/ReplicationTaskProcessorTest.java

@@ -13,6 +13,7 @@
 import static com.netflix.eureka.cluster.TestableInstanceReplicationTask.aReplicationTask;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.*;
 
 /**
  * @author Tomasz Bak
@@ -81,7 +82,7 @@ public void testBatchableTaskCongestionFailureHandling() throws Exception {
         assertThat(status, is(ProcessingResult.Congestion));
         assertThat(task.getProcessingState(), is(ProcessingState.Pending));
     }
-    
+
     @Test
     public void testBatchableTaskNetworkReadTimeOutHandling() throws Exception {
         TestableInstanceReplicationTask task = aReplicationTask().build();
@@ -118,4 +119,13 @@ public void testBatchableTaskPermanentFailureHandling() throws Exception {
         assertThat(status, is(ProcessingResult.Success));
         assertThat(task.getProcessingState(), is(ProcessingState.Failed));
     }
+
+    @Test
+    public void testHandlingNullMessagesWhileEvaluatingTimeout() {
+        assertThatCode(() -> {
+            replicationClient.withEmptyMessageException();
+            TestableInstanceReplicationTask task = aReplicationTask().build();
+            replicationTaskProcessor.process(Collections.singletonList(task));
+        }).doesNotThrowAnyException();
+    }
 }

eureka-core/src/test/java/com/netflix/eureka/cluster/TestableHttpReplicationClient.java

@@ -19,12 +19,14 @@
 import com.netflix.eureka.cluster.protocol.ReplicationList;
 import com.netflix.eureka.cluster.protocol.ReplicationListResponse;
 import com.netflix.eureka.resources.ASGResource.ASGStatus;
+import org.apache.http.client.ClientProtocolException;
+import org.apache.http.client.NonRepeatableRequestException;
 
 import static com.netflix.discovery.shared.transport.EurekaHttpResponse.anEurekaHttpResponse;
 
 /**
  * This stub implementation is primarily useful for batch updates, and complex failure scenarios.
- * Using mock would results in too convoluted code.
+ * Using mock would result in too convoluted code.
  *
  * @author Tomasz Bak
  */
@@ -34,7 +36,7 @@ public class TestableHttpReplicationClient implements HttpReplicationClient {
     private InstanceInfo instanceInfoFromPeer;
     private int networkFailuresRepeatCount;
     private int readtimeOutRepeatCount;
-
+    private boolean nullMessageException;
 
     private int batchStatusCode;
 
@@ -43,7 +45,7 @@ public class TestableHttpReplicationClient implements HttpReplicationClient {
     private final AtomicInteger readTimeOutCounter = new AtomicInteger();
 
     private long processingDelayMs;
-    
+
 
     private final BlockingQueue<HandledRequest> handledRequests = new LinkedBlockingQueue<>();
 
@@ -75,6 +77,10 @@ public HandledRequest nextHandledRequest(long timeout, TimeUnit timeUnit) throws
         return handledRequests.poll(timeout, timeUnit);
     }
 
+    public void withEmptyMessageException() {
+        this.nullMessageException = true;
+    }
+
     @Override
     public EurekaHttpResponse<Void> register(InstanceInfo info) {
         handledRequests.add(new HandledRequest(RequestType.Register, info));
@@ -149,13 +155,17 @@ public EurekaHttpResponse<InstanceInfo> getInstance(String appName, String id) {
 
     @Override
     public EurekaHttpResponse<ReplicationListResponse> submitBatchUpdates(ReplicationList replicationList) {
-
+
+        if(nullMessageException) {
+            throw new RuntimeException(new ClientProtocolException(new NonRepeatableRequestException()));
+        }
+
         if (readTimeOutCounter.get() < readtimeOutRepeatCount) {
             readTimeOutCounter.incrementAndGet();
             throw new RuntimeException(new SocketTimeoutException("Read timed out"));
         }
-    	
-    	
+
+
         if (networkFailureCounter.get() < networkFailuresRepeatCount) {
             networkFailureCounter.incrementAndGet();
             throw new RuntimeException(new IOException("simulated network failure"));