fix: escalate peer read timeouts after max retries — GH#4777

apiml-common/src/main/java/org/zowe/apiml/product/eureka/client/ApimlPeerEurekaNode.java

@@ -43,6 +43,8 @@
 import com.netflix.eureka.util.batcher.TaskDispatchers;
 import com.netflix.eureka.util.batcher.TaskProcessor;
 import lombok.extern.slf4j.Slf4j;
+import org.zowe.apiml.message.log.ApimlLogger;
+import org.zowe.apiml.message.yaml.YamlMessageServiceInstance;
 
 import javax.net.ssl.SSLException;
 import java.io.IOException;
@@ -342,6 +344,11 @@ private static long getLeaseRenewalOf(InstanceInfo info) {
     @Slf4j
     public static class ReplicationTaskProcessor implements TaskProcessor<ReplicationTask> {
 
+        private static final String PEER_REPLICATION_PERMANENT_ERROR = "org.zowe.apiml.common.peerReplicationPermanentError";
+        private static final String PEER_REPLICATION_READ_TIMEOUT = "org.zowe.apiml.common.peerReplicationReadTimeout";
+
+        private static final ApimlLogger apimlLog = ApimlLogger.of(ReplicationTaskProcessor.class, YamlMessageServiceInstance.getInstance());
+
         private final HttpReplicationClient replicationClient;
 
         private final String peerId;
@@ -364,7 +371,7 @@ class NetworkIssueCounter {
 
             final AtomicInteger counter = new AtomicInteger(0);
 
-            private String getCountText() {
+            String getCountText() {
                 int count = counter.get();
 
                 StringBuilder sb = new StringBuilder();
@@ -418,9 +425,7 @@ public ProcessingResult process(ReplicationTask task) {
             } catch (Throwable e) {
                 networkIssueCounter.fail(e.getLocalizedMessage());
                 if (maybeReadTimeOut(e)) {
-                    log.error("It seems to be a socket read timeout exception, it will retry later. if it continues to happen and some eureka node occupied all the cpu time, you should set property 'eureka.server.peer-node-read-timeout-ms' to a bigger value", e);
-                    //read timeout exception is more Congestion than TransientError, return Congestion for longer delay
-                    return ProcessingResult.Congestion;
+                    return handleReadTimeout(e, "The replication task");
                 } else if (isNetworkConnectException(e) && !networkIssueCounter.hasReachedMax()) {
                     logNetworkErrorSample(task, "; retrying after delay.", e);
                     return ProcessingResult.TransientError;
@@ -455,9 +460,7 @@ public ProcessingResult process(List<ReplicationTask> tasks) {
             } catch (Throwable e) {
                 networkIssueCounter.fail(e.getLocalizedMessage());
                 if (maybeReadTimeOut(e)) {
-                    log.error("It seems to be a socket read timeout exception, it will retry later. if it continues to happen and some eureka node occupied all the cpu time, you should set property 'eureka.server.peer-node-read-timeout-ms' to a bigger value", e);
-                    //read timeout exception is more Congestion than TransientError, return Congestion for longer delay
-                    return ProcessingResult.Congestion;
+                    return handleReadTimeout(e, "Batch replication tasks");
                 } else if (isNetworkConnectException(e) && !networkIssueCounter.hasReachedMax()) {
                     logNetworkErrorSample(null, "; retrying after delay.", e);
                     return ProcessingResult.TransientError;
@@ -470,6 +473,17 @@ public ProcessingResult process(List<ReplicationTask> tasks) {
             return ProcessingResult.Success;
         }
 
+        private ProcessingResult handleReadTimeout(Throwable e, String taskDescription) {
+            if (networkIssueCounter.hasReachedMax()) {
+                apimlLog.log(PEER_REPLICATION_PERMANENT_ERROR, networkIssueCounter.getCountText(), taskDescription);
+                return ProcessingResult.PermanentError;
+            }
+            apimlLog.log(PEER_REPLICATION_READ_TIMEOUT, e.getMessage());
+            log.debug("Peer replication socket read timeout", e);
+            //read timeout exception is more Congestion than TransientError, return Congestion for longer delay
+            return ProcessingResult.Congestion;
+        }
+
         /**
          * We want to retry eagerly, but without flooding log file with tons of error entries.
          * As tasks are executed by a pool of threads the error logging multiplies. For example:

apiml-common/src/main/resources/common-log-messages.yml

@@ -34,6 +34,20 @@ messages:
       reason: "Too many concurrent connection requests were made."
       action: "Further connections will be queued until there is room in the connection pool. You may also increase the total connection limit via the gateway start-up script by setting the Gateway configuration for maxTotalConnections."
 
+    - key: org.zowe.apiml.common.peerReplicationPermanentError
+      number: ZWEAO107
+      type: ERROR
+      text: "Socket read timeout has repeatedly reached the maximum retry count (%s). %s will be dropped as a permanent error."
+      reason: "Peer replication did not receive a response before the configured read timeout for the maximum number of retries."
+      action: "Verify the peer Discovery Service is reachable and responsive. If the condition persists, increase eureka.server.peer-node-read-timeout-ms. The peer will catch up through periodic registry synchronization."
+
+    - key: org.zowe.apiml.common.peerReplicationReadTimeout
+      number: ZWEAO108
+      type: ERROR
+      text: "Peer replication socket read timeout occurred: %s. The replication task will be retried later."
+      reason: "The peer Discovery Service did not respond before the configured read timeout."
+      action: "If the timeout continues, verify the peer Discovery Service is reachable and responsive, and consider increasing eureka.server.peer-node-read-timeout-ms."
+
     # HTTP,Protocol messages
     # 400-499
 

apiml-common/src/test/java/org/zowe/apiml/product/eureka/client/ReplicationTaskProcessorTest.java

@@ -18,6 +18,7 @@
 import org.junit.jupiter.api.Test;
 
 import javax.net.ssl.SSLException;
+import java.net.SocketTimeoutException;
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.IntStream;
@@ -130,6 +131,51 @@ void whenNetworkProblemRepeatedMultipleTimes_thenResetCounterAfterSuccessfulConn
             status = replicationTaskProcessor.process(task2);
             assertThat(status, is(ProcessingResult.PermanentError));
         }
+
+        @Test
+        void whenReadTimeoutRepeatedMultipleTimes_thenEscalatesToPermanentError() {
+            TestableInstanceReplicationTask task = aReplicationTask()
+                .withAction(Action.Heartbeat)
+                .withException(new SocketTimeoutException("Read timed out"))
+                .withNetworkFailures(DEFAULT_MAX_RETRIES)
+                .build();
+
+            // First read timeout should cause Congestion
+            ProcessingResult status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.Congestion));
+
+            IntStream.range(1, DEFAULT_MAX_RETRIES - 2).forEach(n -> replicationTaskProcessor.process(task));
+
+            // 9th read timeout should still cause Congestion
+            status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.Congestion));
+
+            // 10th read timeout should escalate to PermanentError
+            status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.PermanentError));
+        }
+
+        @Test
+        void whenConnectionFailureRepeatedMultipleTimes_thenBehaviorUnchanged() {
+            TestableInstanceReplicationTask task = aReplicationTask()
+                .withAction(Action.Heartbeat)
+                .withNetworkFailures(DEFAULT_MAX_RETRIES)
+                .build();
+
+            // First connection failure should cause TransientError
+            ProcessingResult status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.TransientError));
+
+            IntStream.range(1, DEFAULT_MAX_RETRIES - 2).forEach(n -> replicationTaskProcessor.process(task));
+
+            // 9th connection failure should still cause TransientError
+            status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.TransientError));
+
+            // 10th connection failure should cause PermanentError
+            status = replicationTaskProcessor.process(task);
+            assertThat(status, is(ProcessingResult.PermanentError));
+        }
     }
 
     @Nested
@@ -265,5 +311,50 @@ void whenNetworkProblemRepeatedMultipleTimes_thenResetCounterAfterSuccessfulConn
             status = replicationTaskProcessor.process(tasks);
             assertThat(status, is(ProcessingResult.TransientError));
         }
+
+        @Test
+        void whenReadTimeoutRepeatedMultipleTimes_thenEscalatesToPermanentError() {
+            TestableInstanceReplicationTask task = aReplicationTask().build();
+            List<ReplicationTask> tasks = Collections.singletonList(task);
+            replicationClient.withReadtimeOut(DEFAULT_MAX_RETRIES);
+
+            // First read timeout should cause Congestion
+            ProcessingResult status = replicationTaskProcessor.process(tasks);
+            assertThat(status, is(ProcessingResult.Congestion));
+
+            IntStream.range(1, DEFAULT_MAX_RETRIES - 2).forEach(n -> replicationTaskProcessor.process(tasks));
+
+            // 9th read timeout should still cause Congestion
+            status = replicationTaskProcessor.process(tasks);
+            assertThat(status, is(ProcessingResult.Congestion));
+
+            // 10th read timeout should escalate to PermanentError
+            status = replicationTaskProcessor.process(tasks);
+            assertThat(status, is(ProcessingResult.PermanentError));
+        }
+
+        @Test
+        void whenSuccessfulReplicationResetsReadTimeoutCounter() {
+            TestableInstanceReplicationTask task = aReplicationTask().build();
+            List<ReplicationTask> tasks = Collections.singletonList(task);
+
+            // Accumulate 5 read timeouts (not enough to reach max of 10)
+            replicationClient.withReadtimeOut(5);
+            IntStream.range(0, 5).forEach(n -> replicationTaskProcessor.process(tasks));
+
+            // Successful replication should reset the counter
+            replicationClient.withReadtimeOut(0);
+            replicationClient.withBatchReply(200);
+            replicationClient.withNetworkStatusCode(200);
+            ProcessingResult status = replicationTaskProcessor.process(tasks);
+            assertThat(status, is(ProcessingResult.Success));
+
+            // Now 5 more read timeouts should start fresh (Congestion, not PermanentError)
+            replicationClient.withReadtimeOut(10);  // client counter at 5, so 5 more timeouts fire
+            for (int i = 0; i < 5; i++) {
+                status = replicationTaskProcessor.process(tasks);
+                assertThat(status, is(ProcessingResult.Congestion));
+            }
+        }
     }
 }

apiml/src/main/resources/application.yml

@@ -14,7 +14,7 @@ eureka:
     server:
         max-threads-for-peer-replication: 6
         useReadOnlyResponseCache: false
-        peer-node-read-timeout-ms: 15000
+        peer-node-read-timeout-ms: 30000 # Gives peer replication more headroom before permanent-error escalation (GH#4777)
 spring:
     profiles:
         group:

discovery-service/src/main/resources/application.yml

@@ -78,7 +78,7 @@ eureka:
     server:
         max-threads-for-peer-replication: 6
         useReadOnlyResponseCache: false
-        peer-node-read-timeout-ms: 15000
+        peer-node-read-timeout-ms: 30000 # Gives peer replication more headroom before permanent-error escalation (GH#4777)
 
 management:
     endpoints: