KafkaConsumer 在轮询时进入无限等待状态答案

【问题标题】：KafkaConsumer is going into indefinite wait state while pollingKafkaConsumer 在轮询时进入无限等待状态
【发布时间】：2018-05-29 17:16:32
【问题描述】：

我正在尝试使用 KafkaConsumer API 轮询 kafka 主题。但它会进入无限期等待状态，即使我们通过轮询超时也不兑现。

从线程转储显示可运行状态，我进行了多次线程转储，主线程始终保持在同一位置，我相信它不会退出等待。

"main" #1 prio=5 os_prio=0 tid=0x00007f42a800f000 nid=0x59 runnable [0x00007f42b0782000]
   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:79)
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
        - locked <0x00000006c02e2088> (a sun.nio.ch.Util$2)
        - locked <0x00000006c02e2078> (a java.util.Collections$UnmodifiableSet)
        - locked <0x00000006c02e1f60> (a sun.nio.ch.EPollSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at org.apache.kafka.common.network.Selector.select(Selector.java:425)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:270)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:303)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:197)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:187)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:126)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:186)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:857)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:829)
        at org.test.TestReceiver(TestReceiver:100)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:497)
        at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:879)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:197)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:227)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:136)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

这是代码 sn-p..仅打印第一个日志。

LOG.info("Going to wait {}ms for ConsumerRecords", POLLING_TIMEOUT_MILLIS);
ConsumerRecords<String, String> records = consumer.poll(POLLING_TIMEOUT_MILLIS); 
LOG.info("Received {} ConsumerRecords to process.", (records != null ? records.count() : null));

库版本...
kafka_2.11:jar:0.9.0.0
kafka-clients:jar:0.9.0.0

【问题讨论】：

标签： java apache-kafka spark-streaming kafka-consumer-api

【解决方案1】：

KafkaConsumer#poll() 在需要刷新其元数据但无法连接到集群时可能会阻塞。

通过 KIP-266 寻址：https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75974886

【讨论】：

【解决方案2】：

在 kafka 中注意到以下错误，在 server.properties 中配置 offsets.topic.replication.factor=1 后问题得到解决。

ERROR [KafkaApi-0] Number of alive brokers '1' does not meet the required replication factor '3' for the offsets topic (configured via 'offsets.topic.replication.factor'). This error can be ignored if the cluster is starting up and not all brokers are up yet. (kafka.server.KafkaApis)

【讨论】：