Spring Kafka@kafkalistener-重试发送失败的消息并手动提交偏移量

Question

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

spring.kafka.listener.ack-mode=manual-immediate

logging.level.org.springframework.kafka=debug

@KafkaListener(topics = "testTopic",groupId="group-1")
    public void consumerOne(ConsumerRecord<String, String> record,Acknowledgment ack) {
        logger.info("OF PayLoad:{}", record);
        logger.info("OF value:"+record.value());
        logger.info("OF Offset:{}",record.offset());
        ack.acknowledge();

    }
    @KafkaListener(topics = "testTopic",groupId="group-2")
    public void consumerTwo(ConsumerRecord<String, String> record,Acknowledgment ack) {
        logger.info("MF PayLoad_2:{}", record);
        logger.info("MF value_2:"+record.value());
        logger.info("MF Offset_2:{}",record.offset());
        //ack.acknowledge();
    }

在这里，consumerTwo没有确认该消息。我想根据可配置的时间重试将所有消息再次发送给消费者。

任何例子或建议的建议将是很大的帮助。

Answer 1

看这个答案。

在2.2中，我们添加了向seektocurrenterrorhandler添加恢复器的功能，该恢复器可以执行某些操作。

最后一个注释中的pull请求显示了如果恢复器成功恢复消息，如何提交偏移量。这将在周四的2.2.4版本中播出。