You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ce...@apache.org on 2024/09/07 14:24:05 UTC
(kafka) branch trunk updated: MINOR: Improve error message when Connect's EmbeddedKafkaCluster::verifyClusterReadiness method fails (#16918)
This is an automated email from the ASF dual-hosted git repository.
cegerton pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 50e7022a1bf MINOR: Improve error message when Connect's EmbeddedKafkaCluster::verifyClusterReadiness method fails (#16918)
50e7022a1bf is described below
commit 50e7022a1bf756224eeab43579ddb0c3aa03b8be
Author: Chris Egerton <fe...@gmail.com>
AuthorDate: Sat Sep 7 10:23:59 2024 -0400
MINOR: Improve error message when Connect's EmbeddedKafkaCluster::verifyClusterReadiness method fails (#16918)
Reviewers: Greg Harris <gr...@aiven.io>
---
.../util/clusters/EmbeddedKafkaCluster.java | 23 ++++++++++++++++------
1 file changed, 17 insertions(+), 6 deletions(-)
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java
index bfef3e6ae33..aa626af9704 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java
@@ -88,6 +88,7 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_
import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* Setup an embedded Kafka KRaft cluster (using {@link kafka.testkit.KafkaClusterTestKit} internally) with the
@@ -182,14 +183,24 @@ public class EmbeddedKafkaCluster {
Map<String, Object> consumerConfig = Collections.singletonMap(GROUP_ID_CONFIG, consumerGroupId);
String topic = "consumer-warmup-" + consumerGroupId;
- createTopic(topic);
- produce(topic, "warmup message key", "warmup message value");
+ try {
+ createTopic(topic);
+ produce(topic, "warmup message key", "warmup message value");
- try (Consumer<?, ?> consumer = createConsumerAndSubscribeTo(consumerConfig, topic)) {
- ConsumerRecords<?, ?> records = consumer.poll(Duration.ofMillis(GROUP_COORDINATOR_AVAILABILITY_DURATION_MS));
- if (records.isEmpty()) {
- throw new AssertionError("Failed to verify availability of group coordinator and produce/consume APIs on Kafka cluster in time");
+ try (Consumer<?, ?> consumer = createConsumerAndSubscribeTo(consumerConfig, topic)) {
+ ConsumerRecords<?, ?> records = consumer.poll(Duration.ofMillis(GROUP_COORDINATOR_AVAILABILITY_DURATION_MS));
+ if (records.isEmpty()) {
+ throw new AssertionError("Failed to verify availability of group coordinator and/or consume APIs on Kafka cluster in time");
+ }
}
+ } catch (Throwable e) {
+ fail(
+ "The Kafka cluster used in this test was not able to start successfully in time. "
+ + "If no recent changes have altered the behavior of Kafka brokers or clients, and this error "
+ + "is not occurring frequently, it is probably the result of the testing machine being temporarily "
+ + "overloaded and can be safely ignored.",
+ e
+ );
}
try (Admin admin = createAdminClient()) {