gnodet commented on code in PR #22979:
URL: https://github.com/apache/camel/pull/22979#discussion_r3194214236
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaTestUtil.java:
##########
@@ -91,14 +91,13 @@ public static void configureKafkaComponent(CamelContext
context, String bootstra
}
public static void createTopic(KafkaService service, String topic, int
numPartitions) {
- AdminClient kafkaAdminClient = createAdminClient(service);
- NewTopic testTopic = new NewTopic(topic, numPartitions,
CreateTopicsRequest.NO_REPLICATION_FACTOR);
- kafkaAdminClient.createTopics(Collections.singleton(testTopic));
- KafkaFuture<TopicDescription> tdFuture
- =
kafkaAdminClient.describeTopics(Collections.singletonList(topic)).topicNameValues().get(topic);
-
- try {
- TopicDescription td = tdFuture.get(5L, TimeUnit.SECONDS);
+ try (AdminClient kafkaAdminClient = createAdminClient(service)) {
+ NewTopic testTopic = new NewTopic(topic, numPartitions,
CreateTopicsRequest.NO_REPLICATION_FACTOR);
+ kafkaAdminClient.createTopics(Collections.singleton(testTopic));
+ KafkaFuture<TopicDescription> tdFuture
+ =
kafkaAdminClient.describeTopics(Collections.singletonList(topic)).topicNameValues().get(topic);
+
+ TopicDescription td = tdFuture.get(30L, TimeUnit.SECONDS);
Review Comment:
_Claude Code on behalf of Guillaume Nodet_
Good point, you're right that the exception is
`UnknownTopicOrPartitionException` which returns immediately rather than timing
out. I've replaced the simple timeout increase with Awaitility polling: it now
retries `describeTopics` every second for up to 30 seconds, using
`ignoreExceptions()` to treat the `UnknownTopicOrPartitionException` as a "not
yet" signal.
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAutoInstResumeRouteStrategyIT.java:
##########
@@ -54,20 +56,21 @@ public static KafkaResumeStrategyConfigurationBuilder
getDefaultKafkaResumeStrat
public void before() {
Properties props = KafkaTestUtil.getDefaultProperties(service);
KafkaTestUtil.createTopic(service, TOPIC, 1);
- KafkaProducer<Object, Object> producer = new KafkaProducer<>(props);
-
- for (int i = 0; i < 10; i++) {
- producer.send(new ProducerRecord<>(TOPIC, String.valueOf(i)));
+ try (KafkaProducer<Object, Object> producer = new
KafkaProducer<>(props)) {
+ for (int i = 0; i < 10; i++) {
+ producer.send(new ProducerRecord<>(TOPIC, String.valueOf(i)));
+ }
+ producer.flush();
}
}
@Test
@Timeout(value = 30)
- public void testOffsetIsBeingChecked() throws InterruptedException {
+ public void testOffsetIsBeingChecked() {
MockEndpoint mock =
contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
mock.expectedMessageCount(10);
- mock.assertIsSatisfied();
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() ->
mock.assertIsSatisfied());
Review Comment:
_Claude Code on behalf of Guillaume Nodet_
Good catch — lowered the Awaitility timeout to 25 seconds (vs the 30s
`@Timeout`) so the assertion failure message is precise rather than a generic
JUnit timeout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
