apupier commented on code in PR #22979:
URL: https://github.com/apache/camel/pull/22979#discussion_r3194136298
##########
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:
I do not think that chaging the tiemout will change anything. Based on the
stack trace, I understand that the answer is returned as soon as it is possible
and i tis throwing:
```
.UnknownTopicOrPartitionException: This server does not host this
topic-partition.
```
In case, it is a matter of time that the topic are not created, we need to
use an await() and check several times if there are some topics. (which we do
not know if it is a matter of time or if there is a differnt problem that can
prevent the creation of the topics from time to time)
##########
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:
I think we need to be slightly lower than the timeout of the test to benefit
from this assertion to fail with precise message instead of the genric timeout
for the test.
--
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]
