CommitFailedException Commit не может быть завершена, так как группа уже перебалансировала и назначила разделы другому члену

Я использовал kafka 0.10.2 и теперь столкнулся с CommitFailedException. как:

Commit не может быть завершен, так как группа уже перебалансирована и назначил разделы другому члену. Это означает, что время между последующими вызовами poll() было больше, чем сконфигурированное max.poll.interval.ms, что обычно подразумевает, что цикл опроса тратит слишком много времени на обработку сообщений. Вы можете обратиться к этому либо путем увеличения тайм-аута сеанса или уменьшения максимального размера партии, возвращенные в poll() с max.poll.records.

Я установил 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, кажется, что сердцебиение не в тайм-ауте сеанса в моем коде. почему бы не бросить CommitFailedException?

Ответ 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, новый HandleRebalance())

Преимущество этого:

Сообщения не будут повторяться, когда происходит перебалансировка.
Отсутствие исключения сбоя исключение во главе

Ответ 2

session.timeout.ms установленное для потребителя, должно быть меньше значения group.max.session.timeout.ms установленного для брокера Kafka.

Это решило проблему для меня.

Ссылка на ссылку github Commit Failures