RetryingBatchErrorHandler - 偏移提交处理

Question

我正在使用 spring-kafka 2.3.8，我正在尝试记录恢复的记录并使用 RetryingBatchErrorHandler 提交偏移量。您将如何在恢复器中提交偏移量？

public class Customizer implements ContainerCustomizer{
    private static ConsumerRecordRecoverer createConsumerRecordRecoverer() {
        return (consumerRecord, e) -> {
            log.info("Number of attempts exhausted. parition: " consumerRecord.partition() + ", offset: " + consumerRecord.offset());
         # need to commit the offset
         };
    }

    @Override
    public void configure(AbstractMessageListenerContainer container) {
        container.setBatchErrorHandler(new RetryingBatchErrorHandler(new FixedBackOff(5000L, 3L), createConsumerRecordRecoverer()));
    }

Answer 1

如果错误处理程序“处理”异常，容器将自动提交偏移量，除非您将 ackAfterHandle 属性设置为 false（默认为 true）。

编辑

这对我来说按预期工作：

@SpringBootApplication
public class So69534923Application {

    private static final Logger log = LoggerFactory.getLogger(So69534923Application.class);

    public static void main(String[] args) {
        SpringApplication.run(So69534923Application.class, args);
    }

    @KafkaListener(id = "so69534923", topics = "so69534923")
    void listen(List<String> in) {
        System.out.println(in);
        throw new RuntimeException("test");
    }

    @Bean
    RetryingBatchErrorHandler eh() {
        return new RetryingBatchErrorHandler(new FixedBackOff(1000L, 2), (rec, ex) -> {
            this.log.info("Retries exchausted for " + ListenerUtils.recordToString(rec, true));
        });
    }

    @Bean
    ApplicationRunner runner(ConcurrentKafkaListenerContainerFactory<?, ?> factory,
            KafkaTemplate<String, String> template) {

        factory.getContainerProperties().setCommitLogLevel(Level.INFO);
        return args -> {
            template.send("so69534923", "foo");
            template.send("so69534923", "bar");
        };
    }

}

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.type=batch

so69534923: partitions assigned: [so69534923-0]
[foo, bar]
[foo, bar]
[foo, bar]
Retries exchausted for so69534923-0@2
Retries exchausted for so69534923-0@3
Committing: {so69534923-0=OffsetAndMetadata{offset=4, leaderEpoch=null, metadata=''}}

日志来自第二次运行。

EDIT2

它不适用于 2.3.x；您应该升级到受支持的版本。

https://spring.io/projects/spring-kafka#learn