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

Question

我有一个spring kafka消费者，它读取记录并将其交给缓存。计划的任务将定期清除缓存中的记录。我只想在批成功保存到数据库中后更新提交偏移量。我尝试将Acknowledgement对象传递给缓存服务以调用acknowledge方法，如下所示。

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();
        }
    }
}

确认模式设置如下：

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

自动提交是错误的：

config.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false );

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

我正在使用spring-kafka 2.1.7。释放。

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

Answer 1

问题是，消费者线程负责提交偏移量。在轮询时，消费者线程将提交上一批偏移量。

因为在您的情况下，AUTO_COMMIT是错误的，lastAcknowledge。acknowledge（）未确认偏移量未提交。

只有一种方法可以做到这一点，一旦你获得投票记录，将任务安排为异步任务，并在html" target="_blank">异步任务完成后保持消费者线程并提交偏移量，检查此答案作为参考答案

注意：如果你按住消费者线程超过5分钟，将在此处进行重新平衡

新的Java消费者现在支持后台线程的心跳。有一个新的配置max.poll。间隔ms，它控制用户主动离开组之前轮询调用之间的最长时间（默认为5分钟）。配置请求的值。暂停。ms必须始终大于max.poll。间隔ms因为这是消费者重新平衡时JoinGroup请求在服务器上可以阻止的最长时间，所以我们将其默认值更改为略高于5分钟。最后是会话的默认值。暂停。ms已调整为10秒，默认值为max.poll。记录已更改为500。

来自SpringKafka的特别提示

在外国线程上所做的确认将在下一次投票前由消费者线程执行。感谢@Gary Russell提供此信息