listTopics with kafka-clients-2.6.0 gives TimeoutException

Viewed 511

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.

1 Answers

Upgrading kafka-clients to version 2.6.3 worked for me.

Related