我们如何手动重置通过Spring启动java应用程序使用的Kafka主题的偏移量？

Question

我的要求是，当应用程序无法处理从kafka主题的当前偏移量读取的消息时，重置kafka话题的偏移量，该偏移量是通过Spring Bootjava应用程序使用的。手动重置偏移量或发送否定确认后，需要通过spring boot java应用程序的kafka消费者从未提交的偏移量值再次轮询消息。这能实现吗？

Answer 1

这就是Spring Kafka版本的问题

如果您使用的是Spring kafka 2.0.1，则可以使用SeekToCurrentErrorHandler，这将导致在下次轮询中发送失败和未处理的记录。

要使用上述处理程序配置侦听器容器，您需要将其添加到容器属性中

以下是Kafka显示器容器工厂的一个例子

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(this.consumerFactory);
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    return factory;
}

这里是详细的留档：
https://docs.spring.io/spring-kafka/reference/html/_reference.html#_seek_to_current_container_error_handlers