批量插入成功后更新 Kafka 提交偏移量

Question

我有一个 spring-kafka 消费者，它读取记录并将它们交给缓存。计划任务会定期清除缓存中的记录。我只想在批次成功保存在数据库中后更新 COMMIT OFFSET。我尝试将确认对象传递给缓存服务以调用确认方法，如下所示。

public class KafkaConsumer {
    @KafkaListener( topicPattern = "${kafka.topicpattern}", containerFactory = "kafkaListenerContainerFactory" )
    public void receive( ConsumerRecord<String, String> record, Acknowledgment acknowledgment ) {
        cacheService.add( record.value(), acknowledgment );
    }
}

public class CacheService {
    // concurrency handling has been left out in favor of readability
    public void add( String record, Acknowledgment acknowledgment ) {
        this.records.add(record);
        this.lastAcknowledgment = acknowledgment;
    }

    public void saveBatch() { //called by scheduled task
        if( records.size() == BATCH_SIZE ) {
            // perform batch insert into database
            this.lastAcknowledgment.acknowledge();
            this.records.clear();
        }
    }
}

AckMode 设置如下：

factory.getContainerProperties().setAckMode( AbstractMessageListenerContainer.AckMode.MANUAL );

并且自动提交是假的：

config.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false );

即使调用了确认方法，提交偏移量也不会更新。持久化记录后更新提交偏移量的最佳方法是什么？

我正在使用 spring-kafka 2.1.7.RELEASE。

---

编辑：在@GaryRussell confirmed 在下一次轮询期间由消费者线程执行外部线程作出的确认之后，我重新检查了我的代码，发现最后一个确认对象的错误放。 修复此问题后，提交偏移量按预期更新。所以这个问题已经解决了。但是，我无法将此问题标记为已回答。

Answer 1

问题来了，消费者线程负责提交偏移量。在 poll 的时候消费者线程会提交上一个批次的偏移量。

因为在你的情况下AUTO_COMMIT 是假的，lastAcknowledgment.acknowledge() 是不承认偏移量是不提交。

只有一种方法可以做到这一点，一旦您获得轮询记录，将Schedule 任务设为Async 并持有消费者线程并在异步任务完成后提交偏移量，请查看此答案以供参考answer

注意如果您持有消费者线程超过 5 分钟，则会发生重新平衡here

新的 Java Consumer 现在支持来自后台线程的心跳。有一个新的配置 max.poll.interval.ms 控制在消费者主动离开组之前轮询调用之间的最长时间（默认为 5 分钟）。配置 request.timeout.ms 的值必须始终大于max.poll.interval.ms，因为这是消费者重新平衡时 JoinGroup 请求可以在服务器上阻塞的最长时间，因此我们将其默认值更改为略高于 5分钟。最后将session.timeout.ms的默认值调整为10秒，max.poll.records的默认值改为500。

特别说明来自 spring kafka >2.1.5

在下一次投票之前，消费者线程将执行对外部线程的确认感谢@Gary Russell 提供此信息