对于spring-云-流-绑定-Kafka，是否有任何方法可以在最大尝试次数内使用Acknowdge.nack重试？

Question

我正在尝试使用消费批次在Kafka，我发现文件说重试是不支持的，如下所示。

使用批处理模式时不支持在绑定器内重试，因此maxAttempts将被重写为1。您可以配置SeekToCurrentBatchErrorHandler（使用ListenerContainerCustomizer）来实现类似于在绑定器中重试的功能。您还可以使用手动AckMode并调用ACKOWLEDGMENT.NACK（索引，睡眠）来提交部分批处理的偏移量，并重新传递剩余的记录。有关这些技术的更多信息，请参阅Apache Kafka spring文档

如果我使用Acknowdge.nack（索引，睡眠），它会在错误发生时无限重试。是否有任何方法可以在maxAttempts时间内使用Acknowdge.nack重试？

代码就像

@StreamListener(Sink.INPUT)
    public void propagate(@Payload List<PayLoad> payloads, @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
        
        try {
            propagate(payloads);
            acknowledgment.acknowledge();
        } catch (Exception e) {
            acknowledgment.nack(0, 50);
            
        }

    }

Answer 1

没有；你必须自己记录重试次数。

但是，从版本2.5开始，您现在可以使用RecoveringBatchErrorHandler，其中您抛出一个特定的异常来告诉处理程序哪个记录失败了，它会提交该记录之前的记录的偏移量，并对失败的记录应用重试逻辑。

参见https://docs.spring.io/spring-kafka/reference/html/#recovery-batch-eh