Kafka领导人选举导致Kafka流崩溃

Question

我有一个Kafka Streams应用程序，使用3个代理和3个复制因子从Kafka集群进行消费和生产。除了消费者偏移主题（50个分区）之外，所有其他主题都只有一个分区。

当代理尝试首选副本选择时，Streams应用程序（运行在与代理完全不同的实例上）将失败，并出现错误：

Caused by: org.apache.kafka.streams.errors.StreamsException: task [0_0] exception caught when producing
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.checkForException(RecordCollectorImpl.java:119)
    ...
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:197)
Caused by: org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

Streams应用程序尝试成为分区的领导者是否正常，因为它在不属于Kafka集群的服务器上运行？

我可以通过以下方式复制这种行为：

1. 杀死其中一个代理（然后另外2个接管所有分区的领导，按照预期，杀死的代理是他们的领导）
2. 把被杀的经纪人带回来
3. 用bin/kafka-preferred-replica-election.sh--zoo的localhost
触发首选副本领导人选举

我的问题似乎与这个报告的失败相似，所以我想知道这是否是一个新的Kafka Streamsbug。我的完整堆栈跟踪与报告失败中链接的要点完全相同（此处）。

另一个潜在的有趣细节是，在领导者选举期间，我在代理的< code>controller.log中得到这些消息:

[2017-04-12 11:07:50,940] WARN [Controller-3-to-broker-3-send-thread], Controller 3's connection to broker BROKER-3-HOSTNAME:9092 (id: 3 rack: null) was unsuccessful (kafka.controller.RequestSendThread)
java.io.IOException: Connection to BROKER-3-HOSTNAME:9092 (id: 3 rack: null) failed
    at kafka.utils.NetworkClientBlockingOps$.awaitReady$1(NetworkClientBlockingOps.scala:84)
    at kafka.utils.NetworkClientBlockingOps$.blockingReady$extension(NetworkClientBlockingOps.scala:94)
    at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:232)
    at kafka.controller.RequestSendThread.liftedTree1$1(ControllerChannelManager.scala:185)
    at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:184)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

我最初认为这个连接错误是罪魁祸首，但在领导者选举导致 Streams 应用程序崩溃后，如果我重新启动 Streams 应用程序，它会正常工作，直到下一次选举，我根本没有接触经纪人。

所有服务器（3个Kafka代理和Streams应用程序）都在EC2实例上运行。

Answer 1

这个问题现在在0.10.2.1中得到修复。如果您不能选择它，请确保您在streams配置中设置了这两个参数，如下所示:

final Properties props = new Properties();
...
props.put(ProducerConfig.RETRIES_CONFIG, 10);  
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));