当第一次重试抛出的异常与第二次重试不同时，如何重置Spring Kafka消费者中的重试计数？

Question

我正在尝试在 spring-boot 中实现一个 Kafka 重试使用者，并使用 SeekToCurrentErrorHandler 进行重试。

我已将退避策略设置为重试 5 次。

我的问题是，假设第一次重试，异常是“数据库不可用”，第二次尝试 db 可用，但在另一个步骤出现另一个故障，如超时，在这种情况下，重试计数回到零并重新开始，或者将继续尝试仅从第一次重试中剩余的尝试。

我的要求是每次异常与之前抛出的异常不同时将重试计数重置为零。

如何在 Kafka 中实现这一点？

这是我的消费者配置。

@Bean
    public ConcurrentKafkaListenerContainerFactory<String, Anky> kafkaRetryListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, Anky> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);

        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler((record, exception) -> {
            System.out.println(
                    "RetryPolicy** limit has been exceeded! You should really handle this better." + record.key());
        }, new FixedBackOff(0L, 5L));
        errorHandler.addNotRetryableException(IllegalArgumentException.class);
        errorHandler.setCommitRecovered(true);

        factory.setErrorHandler(errorHandler);
        // to keep the consumers alive the failure gets reported to broker so that
        // consumers remain alive
        factory.setStatefulRetry(true);
        factory.setConcurrency(2);
        return factory;
    }

@KafkaListener(topics = "${retry.topic}", groupId = "${consumer.groupId}", containerFactory = "kafkaRetryListenerContainerFactory")
    public void onRetryMessage(ConsumerRecord<String, Anky> record, Acknowledgment acknowledgment)
            throws SystemException {
        LOGGER.debug("are you here in retry?***");
//process message has both DB calls as well as rest calls so it can fail due to a database error and can also fail due to HTTP failure
            processMessage(record, acknowledgment, false, offerIntakeContext);
            }
            
    }

Answer 1

标准错误处理程序无法做到这一点；它不跟踪以前的异常类型。

您需要一个自定义错误处理程序；唯一的其他选择是跟踪侦听器中的先前异常，并在错误处理程序上调用clearThreadState()（来自侦听器方法，在侦听器线程上），这将清除此侦听器线程的所有重试状态。