This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a6f6335fda36 CAMEL-21438: Harden container-exception retry logic and
isolate IggyConsumerOffsetIT (#24508)
a6f6335fda36 is described below
commit a6f6335fda368947c6002f8250f80c4ca4f9cfe9
Author: Adriano Machado <[email protected]>
AuthorDate: Wed Jul 8 04:03:34 2026 -0400
CAMEL-21438: Harden container-exception retry logic and isolate
IggyConsumerOffsetIT (#24508)
Fix isRetryableContainerException short-circuiting on
NoHttpResponseException by
walking the cause chain in a separate pass first, so a dead Docker daemon
fails
fast instead of being retried. Isolate IggyConsumerOffsetIT on dedicated
topic/stream/consumer-group names to eliminate shared-state interference.
Co-Authored-By: Claude Sonnet 5 <[email protected]>
---
.../camel/component/iggy/IggyConsumerOffsetIT.java | 31 ++++++++++++++++++----
.../apache/camel/component/iggy/IggyTestBase.java | 8 ++++--
.../infra/common/services/TestServiceUtil.java | 5 ++++
.../common/services/TestServiceUtilRetryTest.java | 17 ++++++++++++
4 files changed, 54 insertions(+), 7 deletions(-)
diff --git
a/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerOffsetIT.java
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerOffsetIT.java
index 69838091a72a..e784e3bfc7de 100644
---
a/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerOffsetIT.java
+++
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerOffsetIT.java
@@ -16,9 +16,12 @@
*/
package org.apache.camel.component.iggy;
+import java.util.stream.IntStream;
+
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
@@ -26,6 +29,10 @@ import
org.junit.jupiter.api.condition.DisabledIfSystemProperty;
disabledReason = "Iggy 0.6.0+ requires io_uring
which is not available on CI environments")
public class IggyConsumerOffsetIT extends IggyTestBase {
+ private static final String OFFSET_TOPIC = "test-topic-offset";
+ private static final String OFFSET_STREAM = "test-stream-offset";
+ private static final String OFFSET_CONSUMER_GROUP =
"test-consumer-group-offset";
+
int startingOffset = 5;
@Test
@@ -35,10 +42,17 @@ public class IggyConsumerOffsetIT extends IggyTestBase {
mockEndpoint.expectedMessageCount(messages - startingOffset);
for (int i = 0; i < messages; i++) {
- sendMessage("Message " + i);
+ sendMessage(OFFSET_STREAM, OFFSET_TOPIC, "Message " + i);
}
mockEndpoint.assertIsSatisfied();
+ Assertions.assertEquals(
+ IntStream.range(startingOffset, messages)
+ .mapToObj(i -> "Message " + i)
+ .toList(),
+ mockEndpoint.getExchanges().stream()
+ .map(exchange ->
exchange.getMessage().getBody(String.class))
+ .toList());
// send more messages to check offset update
int newMessages = 20;
@@ -46,10 +60,17 @@ public class IggyConsumerOffsetIT extends IggyTestBase {
mockEndpoint.expectedMessageCount(newMessages);
for (int i = 0; i < newMessages; i++) {
- sendMessage("Second Message " + i);
+ sendMessage(OFFSET_STREAM, OFFSET_TOPIC, "Second Message " + i);
}
mockEndpoint.assertIsSatisfied();
+ Assertions.assertEquals(
+ IntStream.range(0, newMessages)
+ .mapToObj(i -> "Second Message " + i)
+ .toList(),
+ mockEndpoint.getExchanges().stream()
+ .map(exchange ->
exchange.getMessage().getBody(String.class))
+ .toList());
}
@Override
@@ -58,13 +79,13 @@ public class IggyConsumerOffsetIT extends IggyTestBase {
@Override
public void configure() {
fromF("iggy:%s?username=%s&password=%s&streamName=%s&host=%s&port=%d&consumerGroupName=%s&autoCommit=false&startingOffset=%d",
- TOPIC,
+ OFFSET_TOPIC,
iggyService.username(),
iggyService.password(),
- STREAM,
+ OFFSET_STREAM,
iggyService.host(),
iggyService.port(),
- CONSUMER_GROUP,
+ OFFSET_CONSUMER_GROUP,
startingOffset)
.to("mock:result");
}
diff --git
a/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyTestBase.java
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyTestBase.java
index 81e20f7d0af1..48d859d81059 100644
---
a/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyTestBase.java
+++
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyTestBase.java
@@ -70,9 +70,13 @@ public abstract class IggyTestBase {
}
protected void sendMessage(String message) {
+ sendMessage(STREAM, TOPIC, message);
+ }
+
+ protected void sendMessage(String stream, String topic, String message) {
client.messages().sendMessages(
- StreamId.of(STREAM),
- TopicId.of(TOPIC),
+ StreamId.of(stream),
+ TopicId.of(topic),
Partitioning.balanced(),
Collections.singletonList(Message.of(message)));
}
diff --git
a/test-infra/camel-test-infra-common/src/main/java/org/apache/camel/test/infra/common/services/TestServiceUtil.java
b/test-infra/camel-test-infra-common/src/main/java/org/apache/camel/test/infra/common/services/TestServiceUtil.java
index 2460c4083a6f..c595c9fea745 100644
---
a/test-infra/camel-test-infra-common/src/main/java/org/apache/camel/test/infra/common/services/TestServiceUtil.java
+++
b/test-infra/camel-test-infra-common/src/main/java/org/apache/camel/test/infra/common/services/TestServiceUtil.java
@@ -68,6 +68,11 @@ public final class TestServiceUtil {
}
private static boolean isRetryableContainerException(Throwable e) {
+ for (Throwable t = e; t != null; t = t.getCause()) {
+ if (t.getClass().getName().contains("NoHttpResponseException")) {
+ return false;
+ }
+ }
for (Throwable t = e; t != null; t = t.getCause()) {
if (t instanceof ContainerFetchException || t instanceof
ContainerLaunchException) {
return true;
diff --git
a/test-infra/camel-test-infra-common/src/test/java/org/apache/camel/test/infra/common/services/TestServiceUtilRetryTest.java
b/test-infra/camel-test-infra-common/src/test/java/org/apache/camel/test/infra/common/services/TestServiceUtilRetryTest.java
index bffdbad7e6a6..9b114576eb8e 100644
---
a/test-infra/camel-test-infra-common/src/test/java/org/apache/camel/test/infra/common/services/TestServiceUtilRetryTest.java
+++
b/test-infra/camel-test-infra-common/src/test/java/org/apache/camel/test/infra/common/services/TestServiceUtilRetryTest.java
@@ -96,6 +96,23 @@ class TestServiceUtilRetryTest {
assertEquals(2, attempts.get());
}
+ @Test
+ void failsImmediatelyOnDockerDaemonNoHttpResponse() throws Exception {
+ AtomicInteger attempts = new AtomicInteger();
+ Throwable noHttpResponse = (Throwable) Class.forName(
+
"com.github.dockerjava.zerodep.shaded.org.apache.hc.core5.http.NoHttpResponseException")
+ .getConstructor(String.class)
+ .newInstance("localhost:2375 failed to respond");
+
+ TestService service = new StubTestService(() -> {
+ attempts.incrementAndGet();
+ throw new ContainerLaunchException("Could not start container",
noHttpResponse);
+ });
+
+ assertThrows(ContainerLaunchException.class, () ->
TestServiceUtil.tryInitialize(service, null));
+ assertEquals(1, attempts.get());
+ }
+
@Test
void succeedsOnFirstAttempt() {
AtomicInteger attempts = new AtomicInteger();
