You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "Guillaume Grossetie (JIRA)" <ji...@apache.org> on 2017/01/18 13:32:26 UTC
[jira] [Updated] (KAFKA-4615) AdminClient.send function poll
without timeout
[ https://issues.apache.org/jira/browse/KAFKA-4615?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Guillaume Grossetie updated KAFKA-4615:
---------------------------------------
Status: Patch Available (was: Open)
> AdminClient.send function poll without timeout
> ----------------------------------------------
>
> Key: KAFKA-4615
> URL: https://issues.apache.org/jira/browse/KAFKA-4615
> Project: Kafka
> Issue Type: Bug
> Components: admin
> Affects Versions: 0.10.1.1
> Environment: Red Hat Enterprise Linux Server release 7.1
> Reporter: Guillaume Grossetie
>
> I'm using the AdminClient to fetch the consumer offsets of my topics.
> When the Kafka cluster is unavailable, I can see the following messages:
> {code}
> 2017-01-11 05:36:45,432 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64444 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:45,569 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64445 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:45,673 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64446 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:45,776 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64447 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:45,880 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64448 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:45,984 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64449 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:46,091 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64450 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> 2017-01-11 05:36:46,198 WARN [kafka-producer-network-thread | producer-1] o.apache.kafka.clients.NetworkClient: Error while fetching metadata with correlation id 64451 : {metrics_app_ida_ido=UNKNOWN_TOPIC_OR_PARTITION}
> // ...
> {code}
> The problem is that the request never ends.
> 4 hours later, my process is stuck and the request is pending.
> Here's a jstack on my Java process :
> {code}
> $ jstack -l 19841
> 2017-01-11 10:32:58
> Full thread dump Java HotSpot(TM) 64-Bit Server VM (25.91-b14 mixed mode):
> "Attach Listener" #12 daemon prio=9 os_prio=0 tid=0x00007f7c5c001000 nid=0x6c0b waiting on condition [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "DestroyJavaVM" #11 prio=5 os_prio=0 tid=0x00007f7c90008800 nid=0x4d88 waiting on condition [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "pool-1-thread-1" #10 prio=5 os_prio=0 tid=0x00007f7c90468800 nid=0x4d96 runnable [0x00007f7c78adc000]
> java.lang.Thread.State: RUNNABLE
> at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
> at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
> at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
> at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
> - locked <0x00000000c52cbc40> (a sun.nio.ch.Util$2)
> - locked <0x00000000c52cbc30> (a java.util.Collections$UnmodifiableSet)
> - locked <0x00000000c52caf88> (a sun.nio.ch.EPollSelectorImpl)
> at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
> at org.apache.kafka.common.network.Selector.select(Selector.java:454)
> at org.apache.kafka.common.network.Selector.poll(Selector.java:277)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:163)
> at kafka.admin.AdminClient.kafka$admin$AdminClient$$send(AdminClient.scala:49)
> at kafka.admin.AdminClient$$anonfun$sendAnyNode$1.apply(AdminClient.scala:61)
> at kafka.admin.AdminClient$$anonfun$sendAnyNode$1.apply(AdminClient.scala:58)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.admin.AdminClient.sendAnyNode(AdminClient.scala:58)
> at kafka.admin.AdminClient.findAllBrokers(AdminClient.scala:87)
> at kafka.admin.AdminClient.listAllGroups(AdminClient.scala:96)
> at kafka.admin.AdminClient.listAllConsumerGroups(AdminClient.scala:111)
> at org.indusbox.idatha.kafka.lag.provider.KafkaLagConsumerOffsetsTopicProvider.fetch(KafkaLagConsumerOffsetsTopicProvider.java:65)
> at org.indusbox.idatha.kafka.lag.KafkaLagPollingReporter.run(KafkaLagPollingReporter.java:49)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
> at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
> at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> Locked ownable synchronizers:
> - <0x00000000c51c7090> (a java.util.concurrent.ThreadPoolExecutor$Worker)
> "kafka-producer-network-thread | producer-1" #9 daemon prio=5 os_prio=0 tid=0x00007f7c9032f800 nid=0x4d95 runnable [0x00007f7c78dde000]
> java.lang.Thread.State: RUNNABLE
> at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
> at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
> at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
> at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
> - locked <0x00000000c51a2a28> (a sun.nio.ch.Util$2)
> - locked <0x00000000c51a2a18> (a java.util.Collections$UnmodifiableSet)
> - locked <0x00000000c51a1bd0> (a sun.nio.ch.EPollSelectorImpl)
> at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
> at org.apache.kafka.common.network.Selector.select(Selector.java:454)
> at org.apache.kafka.common.network.Selector.poll(Selector.java:277)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:229)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:134)
> at java.lang.Thread.run(Thread.java:745)
> Locked ownable synchronizers:
> - None
> "Service Thread" #7 daemon prio=9 os_prio=0 tid=0x00007f7c90140000 nid=0x4d91 runnable [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "C1 CompilerThread1" #6 daemon prio=9 os_prio=0 tid=0x00007f7c9013d800 nid=0x4d90 waiting on condition [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "C2 CompilerThread0" #5 daemon prio=9 os_prio=0 tid=0x00007f7c9013a800 nid=0x4d8f waiting on condition [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "Signal Dispatcher" #4 daemon prio=9 os_prio=0 tid=0x00007f7c90139000 nid=0x4d8e runnable [0x0000000000000000]
> java.lang.Thread.State: RUNNABLE
> Locked ownable synchronizers:
> - None
> "Finalizer" #3 daemon prio=8 os_prio=0 tid=0x00007f7c90105800 nid=0x4d8d in Object.wait() [0x00007f7c7a24e000]
> java.lang.Thread.State: WAITING (on object monitor)
> at java.lang.Object.wait(Native Method)
> - waiting on <0x00000000c52c5168> (a java.lang.ref.ReferenceQueue$Lock)
> at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
> - locked <0x00000000c52c5168> (a java.lang.ref.ReferenceQueue$Lock)
> at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
> at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)
> Locked ownable synchronizers:
> - None
> "Reference Handler" #2 daemon prio=10 os_prio=0 tid=0x00007f7c90101000 nid=0x4d8c in Object.wait() [0x00007f7c801f8000]
> java.lang.Thread.State: WAITING (on object monitor)
> at java.lang.Object.wait(Native Method)
> - waiting on <0x00000000c52c5198> (a java.lang.ref.Reference$Lock)
> at java.lang.Object.wait(Object.java:502)
> at java.lang.ref.Reference.tryHandlePending(Reference.java:191)
> - locked <0x00000000c52c5198> (a java.lang.ref.Reference$Lock)
> at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:153)
> Locked ownable synchronizers:
> - None
> "VM Thread" os_prio=0 tid=0x00007f7c900f9800 nid=0x4d8b runnable
> "GC task thread#0 (ParallelGC)" os_prio=0 tid=0x00007f7c9001d800 nid=0x4d89 runnable
> "GC task thread#1 (ParallelGC)" os_prio=0 tid=0x00007f7c9001f800 nid=0x4d8a runnable
> "VM Periodic Task Thread" os_prio=0 tid=0x00007f7c90142800 nid=0x4d92 waiting on condition
> JNI global references: 287
> {code}
> As you can my thread is still running {{org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)}}
> I think we should use the poll method with a timeout and give up after a short period of time :
> {code}public boolean poll(RequestFuture<?> future, long timeout){code}
> Ref: https://github.com/apache/kafka/blob/75469a3b602c26ea81d6fc0a409d39d321195ea4/core/src/main/scala/kafka/admin/AdminClient.scala#L46
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)