This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-spring-boot.git
The following commit(s) were added to refs/heads/main by this push:
new 6826529ca36 Fix flaky Kafka ITs on Camel-Spring-Boot Daily #397/#398
6826529ca36 is described below
commit 6826529ca36708f4e2edd439a2f7684ed3dcf5d5
Author: croway <[email protected]>
AuthorDate: Tue Oct 13 10:33:01 2026 +0200
Fix flaky Kafka ITs on Camel-Spring-Boot Daily #397/#398
MockEndpoint.assertIsSatisfied(long) only applies the timeout to endpoints
expecting no messages; otherwise it waits the default 10s. Use the static
MockEndpoint.assertIsSatisfied(timeout, unit, endpoints) so the intended
timeout is actually applied in KafkaShareConsumerIT and KafkaConsumerFullIT.
Also port the CAMEL-19358 flakiness fix from apache/camel to
KafkaConsumerFullIT: unique topic and route names, and wait for the topic
deletion to complete in @AfterEach.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../share/springboot/KafkaShareConsumerIT.java | 2 +-
.../kafka/integration/KafkaConsumerFullIT.java | 36 ++++++++++++++--------
2 files changed, 24 insertions(+), 14 deletions(-)
diff --git
a/components-starter/camel-kafka-share-starter/src/test/java/org/apache/camel/component/kafka/share/springboot/KafkaShareConsumerIT.java
b/components-starter/camel-kafka-share-starter/src/test/java/org/apache/camel/component/kafka/share/springboot/KafkaShareConsumerIT.java
index 1fc680b27dc..4616918a5f3 100644
---
a/components-starter/camel-kafka-share-starter/src/test/java/org/apache/camel/component/kafka/share/springboot/KafkaShareConsumerIT.java
+++
b/components-starter/camel-kafka-share-starter/src/test/java/org/apache/camel/component/kafka/share/springboot/KafkaShareConsumerIT.java
@@ -114,7 +114,7 @@ class KafkaShareConsumerIT {
produce(WORK_TOPIC, 5);
- work.assertIsSatisfied(60000);
+ MockEndpoint.assertIsSatisfied(60, TimeUnit.SECONDS, work);
}
@Test
diff --git
a/components-starter/camel-kafka-starter/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
b/components-starter/camel-kafka-starter/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
index 1f8e336362c..572e6d3728b 100644
---
a/components-starter/camel-kafka-starter/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
+++
b/components-starter/camel-kafka-starter/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
@@ -19,6 +19,8 @@ package org.apache.camel.component.kafka.integration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.stream.StreamSupport;
import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
@@ -30,6 +32,7 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.spring.boot.CamelAutoConfiguration;
import org.apache.camel.test.spring.junit6.CamelSpringBootTest;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -57,7 +60,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
KafkaConsumerFullIT.class,
KafkaConsumerFullIT.TestConfiguration.class, })
@DisabledIfSystemProperty(named = "ci.env.name", matches = "github.com",
disabledReason = "Disabled on GH Action due to Docker limit")
public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
- public static final String TOPIC = "test-full";
+ public static final String TOPIC = "test-full-" + Uuid.randomUuid();
+ public static final String ROUTE = "full-it-" + Uuid.randomUuid();
private static final Logger LOG =
LoggerFactory.getLogger(KafkaConsumerFullIT.class);
@@ -79,12 +83,18 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
}
@AfterEach
- public void after() {
+ public void after() throws Exception {
if (producer != null) {
producer.close();
}
// clean all test topics
- kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
+ // wait for the topic to be actually deleted, so the next test does
not race with it
+ try {
+
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all().get(60,
TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ // the topic may not exist if the test did not produce any record
+ LOG.debug("Unable to delete topic {}: {}", TOPIC, e.getMessage());
+ }
to.reset();
}
@@ -110,7 +120,7 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
producer.send(data);
}
- to.assertIsSatisfied(3000);
+ MockEndpoint.assertIsSatisfied(30, TimeUnit.SECONDS, to);
assertEquals(5, StreamSupport
.stream(MockConsumerInterceptor.recordsCaptured.get(0).records(TOPIC).spliterator(),
false).count());
@@ -131,7 +141,7 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
data.headers().add(new RecordHeader(propagatedHeaderKey,
propagatedHeaderValue));
producer.send(data);
- to.assertIsSatisfied(3000);
+ MockEndpoint.assertIsSatisfied(30, TimeUnit.SECONDS, to);
Map<String, Object> headers =
to.getExchanges().get(0).getIn().getHeaders();
assertTrue(headers.containsKey(KafkaConstants.TOPIC),
@@ -149,7 +159,7 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC,
"1", msg);
producer.send(data);
}
- to.assertIsSatisfied(3000);
+ MockEndpoint.assertIsSatisfied(30, TimeUnit.SECONDS, to);
to.reset();
@@ -158,15 +168,15 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1",
"message-2", "message-3", "message-4");
// Restart endpoint,
- context.getRouteController().stopRoute("full-it");
+ context.getRouteController().stopRoute(ROUTE);
KafkaEndpoint kafkaEndpoint = (KafkaEndpoint)
context.getEndpoint(from);
kafkaEndpoint.getConfiguration().setSeekTo(SeekPolicy.BEGINNING);
- context.getRouteController().startRoute("full-it");
+ context.getRouteController().startRoute(ROUTE);
// As wee set seek to beginning we should re-consume all messages
- to.assertIsSatisfied(3000);
+ MockEndpoint.assertIsSatisfied(30, TimeUnit.SECONDS, to);
}
@Order(4)
@@ -179,19 +189,19 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC,
"1", msg);
producer.send(data);
}
- to.assertIsSatisfied(3000);
+ MockEndpoint.assertIsSatisfied(30, TimeUnit.SECONDS, to);
to.reset();
to.expectedMessageCount(0);
// Restart endpoint,
- context.getRouteController().stopRoute("full-it");
+ context.getRouteController().stopRoute(ROUTE);
KafkaEndpoint kafkaEndpoint = (KafkaEndpoint)
context.getEndpoint(from);
kafkaEndpoint.getConfiguration().setSeekTo(SeekPolicy.END);
- context.getRouteController().startRoute("full-it");
+ context.getRouteController().startRoute(ROUTE);
to.assertIsSatisfied(3000);
}
@@ -213,7 +223,7 @@ public class KafkaConsumerFullIT extends
BaseEmbeddedKafkaTestSupport {
public void configure() {
from(from).process(
exchange -> LOG.trace("Captured on the processor:
{}", exchange.getMessage().getBody()))
- .routeId("full-it").to(to);
+ .routeId(ROUTE).to(to);
}
};
}
