diff --git a/apiml-common/src/main/java/org/zowe/apiml/product/eureka/client/ApimlPeerEurekaNode.java b/apiml-common/src/main/java/org/zowe/apiml/product/eureka/client/ApimlPeerEurekaNode.java index 2f92f8e22f..686143b488 100644 --- a/apiml-common/src/main/java/org/zowe/apiml/product/eureka/client/ApimlPeerEurekaNode.java +++ b/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 { + 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 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 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: diff --git a/apiml-common/src/main/resources/common-log-messages.yml b/apiml-common/src/main/resources/common-log-messages.yml index fe7fe4c6be..27bc7e40e9 100644 --- a/apiml-common/src/main/resources/common-log-messages.yml +++ b/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 diff --git a/apiml-common/src/test/java/org/zowe/apiml/product/eureka/client/ReplicationTaskProcessorTest.java b/apiml-common/src/test/java/org/zowe/apiml/product/eureka/client/ReplicationTaskProcessorTest.java index 496d68a430..642078e06b 100644 --- a/apiml-common/src/test/java/org/zowe/apiml/product/eureka/client/ReplicationTaskProcessorTest.java +++ b/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 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 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)); + } + } } } diff --git a/apiml/src/main/resources/application.yml b/apiml/src/main/resources/application.yml index 73407fa023..726dc1dc5a 100644 --- a/apiml/src/main/resources/application.yml +++ b/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: diff --git a/discovery-service/src/main/resources/application.yml b/discovery-service/src/main/resources/application.yml index 7f64fd6b1b..23e020306c 100644 --- a/discovery-service/src/main/resources/application.yml +++ b/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: