我对承认的原则感到困惑

Question

版本:2.1.11

问题：我有一些错误的参数配置:max.poll.records:500 max.poll.interval.ms:10000。但消耗大约需要25000秒。因此，它将遇到异常：org.apache.kafka.clients.consumer.CommitFailedException：提交无法完成，因为组已经重新平衡并将分区分配给了另一个成员。这意味着对poll（）的后续调用之间的时间比配置的max.poll.interval.ms长，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加会话超时，或者通过使用max.poll.records减小poll（）中返回的批的最大大小来解决这一问题。

然后我修正了参数，这样这个错误就解决了。

但是，我很困惑，我已经提交了由ack.acknowledge（）消费的一条消息。为什么偏移量根本不能提交？这是代码

@KafkaListener(topics = "${kafka.topic}")
public void consume(ConsumerRecord<String, SyncResMessage> record, Acknowledgment ack) {
    
    try {
        // consume message
        consumerService.dealResource(record.value());
    } catch (Exception e) {
        LOGGER.error("error when consume data, data key is {}, exception is {}.", record.key(), e);
    }

    if (ack != null) {
        LOGGER.info("commit successfully.");
        ack.acknowledge();
    } else {
        LOGGER.error("message of commit is null, record is {}.", JsonUtil.toString(record));
    }
}

任何帮助都将不胜感激~

Answer 1

也许这就是答案。手动表示不立即提交，它将更新此映射。MANUAL_IMMEDIATE表示立即提交。