apache-kafka - 卡夫卡流 - CommitFailedException : Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member

Question

我正在运行一个 Kafka Stream 应用程序，它使用来自 2 个主题的数据并将加入/合并的结果输出到 3 个主题中。
kafka 主题有 15 个分区和 3 个复制因子。我们有 5 个 kafka 经纪人和 5 个动物园管理员。
我正在运行 15 个 Kafka Stream 应用程序实例，因此每个应用程序可以有 1 个分区。
卡夫卡版本 - 0.11.0.0

我在我的 kafka 流应用程序中遇到以下异常:

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records. at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:725) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:604) at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1173) at org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:307) at org.apache.kafka.streams.processor.internals.StreamTask.access$000(StreamTask.java:49) at org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:268) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187) at org.apache.kafka.streams.processor.internals.StreamTask.commitImpl(StreamTask.java:259) at org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:362) at org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:346) at org.apache.kafka.streams.processor.internals.StreamThread$3.apply(StreamThread.java:1118) at org.apache.kafka.streams.processor.internals.StreamThread.performOnStreamTasks(StreamThread.java:1448) at org.apache.kafka.streams.processor.internals.StreamThread.suspendTasksAndState(StreamThread.java:1110) at org.apache.kafka.streams.processor.internals.StreamThread.access$1800(StreamThread.java:73) at org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsRevoked(StreamThread.java:218) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinPrepare(ConsumerCoordinator.java:422) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:353) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043) at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:582) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:553) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527) 2017-08-09 14:50:49 - [ERROR] [click-live-StreamThread-1] [org.apache.kafka.streams.processor.internals.StreamThread.performOnStreamTasks:1453] :

有人可以请帮助可能是什么原因和解决方案吗？

另外，当我的 kafka 代理中的 1 个关闭时，我的 kafka 流应用程序没有连接到其他代理？
我已经设置了 brokers.list=broker1:9092,broker2:9092,broker3:9092,broker4:9092,broker5:9092

Answer 1

根据信息，这是最有可能的解决途径:
尝试按照消息中的建议进行操作:
“您可以通过增加 session 超时或通过使用 max.poll.records 减少 poll() 中返回的批次的最大大小来解决此问题”