CommitFailedException提交无法完成，因为组已重新平衡并将分区分配给另一个成员

Question

我使用的是Kafka0.10.2，现在面临一个CommitFailedException。如：

无法完成提交，因为组已重新平衡并将分区分配给另一个成员。这意味着对poll（）的后续调用之间的时间比配置的max.poll.interval.ms长，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加会话超时，或者通过使用max.poll.records减小poll（）中返回的批的最大大小来解决这一问题。

我已经将max.poll.interval.ms设置为integer.max_value。那么有谁能告诉我为什么即使我已经设置了值，这种情况还是会发生吗？

另一个问题是：我按照描述将session.timeout.ms设置为60000，但它仍然会发生。我试图通过一个简单的代码来复制

 public static void main(String[] args) throws InterruptedException {     
        Logger logger = Logger.getLogger(KafkaConsumer10.class);
        logger.info("XX");
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka-broker:9098");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("max.poll.interval.ms", "300000");
        props.put("session.timeout.ms", "10000");
        props.put("max.poll.records", "2");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("t1"));
        while (true) {
            Thread.sleep(11000);
            ConsumerRecords<String, String> records = consumer.poll(100);
            //Thread.sleep(11000);
            Thread.sleep(11000);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }

当我将session.timeout.ms设置为10000时，我尝试在轮询循环中睡眠超过10000毫秒，但似乎工作正常，没有例外。所以我对此感到困惑。如果heartbeat是由consumer.poll和consumer.commit触发的，那么在我的代码中heartbeat似乎超出了会话超时。为什么不抛出CommitFailedException？

Answer 1

嗨，为此，您需要在代码中处理重新平衡条件，并且应该处理正在进行的消息并在重新平衡之前提交它

如：

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Implement what you want to do once rebalancing is done.
    }

    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // commit current method
    }
}

并使用以下语法订阅主题：

KafkaConsumer.subscribe（topicNameList,new HandleRebalance（））

这样做的好处是：

>

在重新平衡时不会重复消息。

无提交失败异常

Answer 2

在使用者上设置的session.timeout.ms应小于在Kafka Broker上设置的group.max.session.timeout.ms。

这为我解决了问题。

Credit to github链接提交失败