Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
balhar-jakub wants to merge 2 commits into
base: v3.x.x
Choose a base branch
from
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6021e16
fix: escalate read timeout to PermanentError after max retries (#4777)
balhar-jakub Jul 7, 2026
ef99cc3
Merge branch 'v3.x.x' into hermes/gh4777
balhar-jakub Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion apiml-common/src/main/java/org/zowe/apiml/product/eureka/client/ApimlPeerEurekaNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -364,7 +364,7 @@ class NetworkIssueCounter {

final AtomicInteger counter = new AtomicInteger(0);

private String getCountText() {
String getCountText() {
int count = counter.get();

StringBuilder sb = new StringBuilder();
Expand Down Expand Up @@ -418,6 +418,10 @@ public ProcessingResult process(ReplicationTask task) {
} catch (Throwable e) {
networkIssueCounter.fail(e.getLocalizedMessage());
if (maybeReadTimeOut(e)) {
if (networkIssueCounter.hasReachedMax()) {
log.error("Socket read timeout has occurred repeatedly and reached the maximum retry count ({}). The replication task will be dropped as a permanent error and the peer will catch up via periodic registry sync.", networkIssueCounter.getCountText());
return ProcessingResult.PermanentError;
}
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;
Expand Down Expand Up @@ -455,6 +459,10 @@ public ProcessingResult process(List<ReplicationTask> tasks) {
} catch (Throwable e) {
networkIssueCounter.fail(e.getLocalizedMessage());
if (maybeReadTimeOut(e)) {
if (networkIssueCounter.hasReachedMax()) {
log.error("Socket read timeout has occurred repeatedly and reached the maximum retry count ({}). Batch replication tasks will be dropped as a permanent error and the peer will catch up via periodic registry sync.", networkIssueCounter.getCountText());

@achmelo achmelo Jul 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these 2 log messages are too similar and it would make it harder to debug. I would rephrase one of them and maybe also print the exception message, at least in the debug level

@balhar-jakub balhar-jakub Jul 13, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we use it ourselves, wouldn't it make sense to assign it the standard log id according to standard? Even for the original one.

I will change the text so it is further away as well.

return ProcessingResult.PermanentError;
}
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;
Expand Down
91 changes: 91 additions & 0 deletions ...mmon/src/test/java/org/zowe/apiml/product/eureka/client/ReplicationTaskProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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));
}
}
}
}
2 changes: 1 addition & 1 deletion apiml/src/main/resources/application.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ eureka:
server:
max-threads-for-peer-replication: 6
useReadOnlyResponseCache: false
peer-node-read-timeout-ms: 15000
peer-node-read-timeout-ms: 30000
spring:
cloud:
gateway:
Expand Down
2 changes: 1 addition & 1 deletion discovery-service/src/main/resources/application.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ eureka:
server:
max-threads-for-peer-replication: 6
useReadOnlyResponseCache: false
peer-node-read-timeout-ms: 15000
peer-node-read-timeout-ms: 30000

management:
endpoints:
Expand Down
Loading