KAFKA-16905: Fixed potential thread block by admin API

[apoorvmittal10]

The PR moves the blocking call of describeCluster to async. Though the issue coldn't be re-produced in Kafka intgeration tests but a thread block on describeCluster has been observed when using admin describeTopic API in kafka-rest integrations tests.

The threads blocks indefinetly

"kafka-admin-client-thread | adminclient-3" #114 daemon prio=5 os_prio=31 cpu=6.57ms elapsed=417.17s tid=0x00000001364fc200 nid=0x13403 waiting on condition  [0x00000002bb419000]
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(java.base@17.0.7/Native Method)
	- parking to wait for  <0x0000000773804828> (a java.util.concurrent.CompletableFuture$Signaller)
	at java.util.concurrent.locks.LockSupport.park(java.base@17.0.7/LockSupport.java:211)
	at java.util.concurrent.CompletableFuture$Signaller.block(java.base@17.0.7/CompletableFuture.java:1864)
	at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@17.0.7/ForkJoinPool.java:3463)
	at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.7/ForkJoinPool.java:3434)
	at java.util.concurrent.CompletableFuture.waitingGet(java.base@17.0.7/CompletableFuture.java:1898)
	at java.util.concurrent.CompletableFuture.get(java.base@17.0.7/CompletableFuture.java:2072)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
	at org.apache.kafka.clients.admin.KafkaAdminClient.handleDescribeTopicsByNamesWithDescribeTopicPartitionsApi(KafkaAdminClient.java:2324)
	at org.apache.kafka.clients.admin.KafkaAdminClient.describeTopics(KafkaAdminClient.java:2122)
	at org.apache.kafka.clients.admin.Admin.describeTopics(Admin.java:311)
	at io.confluent.kafkarest.controllers.TopicManagerImpl.describeTopics(TopicManagerImpl.java:155)
	at io.confluent.kafkarest.controllers.TopicManagerImpl.lambda$listTopics$2(TopicManagerImpl.java:76)
	at io.confluent.kafkarest.controllers.TopicManagerImpl$$Lambda$1925/0x0000000800891448.apply(Unknown Source)
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(java.base@17.0.7/CompletableFuture.java:1150)
	at java.util.concurrent.CompletableFuture.postComplete(java.base@17.0.7/CompletableFuture.java:510)
	at java.util.concurrent.CompletableFuture.complete(java.base@17.0.7/CompletableFuture.java:2147)
	at io.confluent.kafkarest.common.KafkaFutures.lambda$toCompletableFuture$0(KafkaFutures.java:45)
	at io.confluent.kafkarest.common.KafkaFutures$$Lambda$1909/0x0000000800897528.accept(Unknown Source)
	at org.apache.kafka.common.internals.KafkaFutureImpl.lambda$whenComplete$2(KafkaFutureImpl.java:107)
	at org.apache.kafka.common.internals.KafkaFutureImpl$$Lambda$1910/0x0000000800897750.accept(Unknown Source)
	at java.util.concurrent.CompletableFuture.uniWhenComplete(java.base@17.0.7/CompletableFuture.java:863)
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(java.base@17.0.7/CompletableFuture.java:841)
	at java.util.concurrent.CompletableFuture.postComplete(java.base@17.0.7/CompletableFuture.java:510)
	at java.util.concurrent.CompletableFuture.complete(java.base@17.0.7/CompletableFuture.java:2147)
	at org.apache.kafka.common.internals.KafkaCompletableFuture.kafkaComplete(KafkaCompletableFuture.java:39)
	at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:122)
	at org.apache.kafka.clients.admin.KafkaAdminClient$4.handleResponse(KafkaAdminClient.java:2106)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1370)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1523)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1446)
	at java.lang.Thread.run(java.base@17.0.7/Thread.java:833)

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[AndrewJSchofield]

Thanks for the PR. Looks good to me.

[omkreddy]

there is checkstyle error

[apoorvmittal10]

there is checkstyle error

My bad, fixed.

[mimaison]

Can you open a Jira so it's easier for users to find? Thanks

[apoorvmittal10]

Can you open a Jira so it's easier for users to find? Thanks

Done, https://issues.apache.org/jira/browse/KAFKA-16905

[lianetm]

Thanks @apoorvmittal10, LGTM.

[mumrah]

Thanks for finding and fix this @apoorvmittal10 👍

LGTM

[ableegoldman]

Hey guys -- just a gentle reminder to please check the builds carefully before merging. This patch seems to have introduced an issue that caused several tests to hang indefinitely, as you can see in the PR build for this which aborted after 7+ hours

We already have a fix and are just waiting to confirm it fixes the issue so no harm done (I've definitely merged something that broke the build myself in the past). Just wanted to raise this since the 3.8 code freeze is fast approaching 🙂