I have a small code to check if a particular topic is already present in Kafka. It worked fine with kafka-clients-2.5.0. But after upgrading to kafka-clients to 2.6.0, it started giving TimeoutException.
This was my original code.
Properties adminProperties = new Properties();
adminProperties.put(ProducerConfig."bootstrap.servers", "localhost:9092");
AdminClient adminClient = KafkaAdminClient.create(adminProperties);
boolean topicExists = adminClient.listTopics().names().get().contains("myDataTopic");
For troubleshooting, I have splitted it and tried extending some timeout values like below. But no use. It works fine with 2.5.1 but not with 2.6.0.
Properties adminProperties = new Properties();
adminProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
adminProperties.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "900000");
AdminClient adminClient = KafkaAdminClient.create(adminProperties);
System.out.println("createKafkaTopic(): Listing Topics...");
ListTopicsResult listTopicsResult = adminClient.listTopics(new ListTopicsOptions().timeoutMs(900000));
System.out.println("createKafkaTopic(): Retrieve Topic names...");
KafkaFuture<Collection<TopicListing>> setKafkaFuture = listTopicsResult.listings();
System.out.println("createKafkaTopic(): Display existing Topics...");
while(!setKafkaFuture.isDone()) {
System.out.println("Waiting...");
Thread.sleep(10);
}
Collection<TopicListing> topicNames = setKafkaFuture.get(900,TimeUnit.SECONDS);
System.out.println(topicNames);
System.out.println("createKafkaTopic(): Check if Topic exists...");
boolean topicExists = topicNames.contains("myDataTopic");
Here is my output:
createKafkaTopic(): Listing Topics...
createKafkaTopic(): Retrieve Topic names...
createKafkaTopic(): Display existing Topics...
Waiting...
Waiting...
Waiting...
Waiting...
Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1604903438670, tries=1, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 1 attempt(s)
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:104)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:272)
at KafkaUtil.createKafkaTopic(KafkaUtil.java:45)
at KafkaUtil.main(KafkaUtil.java:21)
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1604903438670, tries=1, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited.
I saw a similar issue here (How to display topics using Kafka Clients in Java?). But it seems resolved by adding some dependencies. I too tried adding all dependencies to my pom.xml, and no luck.