You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Kyle Kingsbury (Jira)" <ji...@apache.org> on 2024/10/08 20:00:00 UTC
[jira] [Created] (KAFKA-17734) KafkaConsumer.close(0) can block indefinitely in ConsumerNetworkClient.poll
Kyle Kingsbury created KAFKA-17734:
--------------------------------------
Summary: KafkaConsumer.close(0) can block indefinitely in ConsumerNetworkClient.poll
Key: KAFKA-17734
URL: https://issues.apache.org/jira/browse/KAFKA-17734
Project: Kafka
Issue Type: Bug
Components: clients, consumer
Affects Versions: 3.8.0
Environment: openjdk version "17.0.12" 2024-07-16
OpenJDK Runtime Environment (build 17.0.12+7-Ubuntu-1ubuntu222.04)
OpenJDK 64-Bit Server VM (build 17.0.12+7-Ubuntu-1ubuntu222.04, mixed mode, sharing)
Reporter: Kyle Kingsbury
This might be related to KAFKA-17263, but I've got a slightly different stacktrace.
With the official Java Kafka client library, version 3.8.0, calls to `consumer.close()` can stall indefinitely even though I've provided a zero-second timeout. I've tried spawning a separate thread which calls consumer.wakeup() after one second just to make sure the client gets woken up, and that doesn't seem to work either. Jstack indicates the thread calling consumer.close() is stuck on AbstractCoordinator.maybeLeaveGroup -> ConsumerNetworkClient.poll, which in turn is stuck on a future completion handler, which goes *back* into the AbstractCoordinator, which in turn hits onLeaderElected, which eventually issues ConsumerNetworkClient.poll(), and that looks to ground out in a blocking `select()` call.
It would be really nice if there were a way to reliably release client resources--threadpools, network connections, etc--that didn't do in-line network IO.
{{"jepsen worker 24" #24101 prio=5 os_prio=0 cpu=6680.43ms elapsed=10760.37s tid=0x00007462200059e0 nid=0x4909a runnable [0x00007474133fd000]
java.lang.Thread.State: RUNNABLE
at sun.nio.ch.EPoll.wait(java.base@17.0.12/Native Method)
at sun.nio.ch.EPollSelectorImpl.doSelect(java.base@17.0.12/EPollSelectorImpl.java:118)
at sun.nio.ch.SelectorImpl.lockAndDoSelect(java.base@17.0.12/SelectorImpl.java:129)
- locked <0x00000002dcbe20d0> (a sun.nio.ch.Util$2)
- locked <0x00000002dcbe2080> (a sun.nio.ch.EPollSelectorImpl)
at sun.nio.ch.SelectorImpl.select(java.base@17.0.12/SelectorImpl.java:141)
at org.apache.kafka.common.network.Selector.select(Selector.java:878)
at org.apache.kafka.common.network.Selector.poll(Selector.java:469)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:595)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:252)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:243)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:165)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.ensureFreshMetadata(ConsumerNetworkClient.java:176)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.updateGroupSubscription(ConsumerCoordinator.java:557)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeaderElected(ConsumerCoordinator.java:636)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onLeaderElected(AbstractCoordinator.java:766)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$1100(AbstractCoordinator.java:117)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:670)
- locked <0x00000002dcbe2228> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:631)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1310)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1285)
at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:206)
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:169)
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:129)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:616)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:428)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:313)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:322)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1181)
- locked <0x00000002dcbe2228> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:1131)
- locked <0x00000002dcbe2228> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:986)
at org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer.lambda$close$3(LegacyKafkaConsumer.java:1135)
at org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer$$Lambda$466/0x0000747234d46198.run(Unknown Source)
at org.apache.kafka.common.utils.Utils.swallow(Utils.java:1042)
at org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer.close(LegacyKafkaConsumer.java:1135)
at org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer.close(LegacyKafkaConsumer.java:1104)
at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1777)
at jepsen.redpanda.client$close_consumer_BANG_$fn__7227.invoke(client.clj:212)
at jepsen.redpanda.client$close_consumer_BANG_.invokeStatic(client.clj:211)
at jepsen.redpanda.client$close_consumer_BANG_.invoke(client.clj:200)
at jepsen.redpanda.workload.queue.Client.close_BANG_(queue.clj:782)
at jepsen.client.Validate.close_BANG_(client.clj:81)
at jepsen.generator.interpreter.ClientWorker.close_BANG_(interpreter.clj:69)
at jepsen.generator.interpreter.ClientWorker.invoke_BANG_(interpreter.clj:47)
at jepsen.generator.interpreter$spawn_worker$fn__13745$fn__13746.invoke(interpreter.clj:140)
at jepsen.generator.interpreter$spawn_worker$fn__13745.invoke(interpreter.clj:123)
at clojure.core$binding_conveyor_fn$fn__5842.invoke(core.clj:2047)
at clojure.lang.AFn.call(AFn.java:18)
at java.util.concurrent.FutureTask.run(java.base@17.0.12/FutureTask.java:264)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.12/ThreadPoolExecutor.java:1136)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.12/ThreadPoolExecutor.java:635)
at java.lang.Thread.run(java.base@17.0.12/Thread.java:840)}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)