Kafka消费者错误处理：混淆错误处理/RetryTemplate

Question

我尝试在使用邮件时进行以下错误处理：

• 如果出现序列化错误：在DLT中发送消息

我拥有的（2.5.1Kafka客户端的Spring kafka 2.5.5版本）如下：

@Bean(name = "kafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, MyObject> kafkaListenerContainerFactory() throws ExceptionTechnique {
        ConcurrentKafkaListenerContainerFactory<String, MyObject> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(RECORD);
        factory.setRetryTemplate(retryTemplate());
        factory.setRecoveryCallback(context -> {
           Object record = context.getAttribute(RetryingMessageListenerAdapter.CONTEXT_RECORD);
           LOGGER_TECHNIQUE.error("Fail to handle message after {} retries. {}",
                context.getRetryCount(), record);
           return record;
        });
        factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(kafkaOperations,
                        (cr, e) -> new TopicPartition("myTopicToRead.dlq", cr.partition())),
                        new FixedBackOff(10000L, 2L)));
        factory.setConcurrency(kafkaListenerConcurrency);
        factory.setStatefulRetry(true);

        return factory;
    }

@Bean
public RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(backOffPolicy());
    retryTemplate.setRetryPolicy(retryPolicy());
    return retryTemplate;
}

private BackOffPolicy backOffPolicy() {
    FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
    fixedBackOffPolicy.setBackOffPeriod(10000);
    return fixedBackOffPolicy;
}

private SimpleRetryPolicy retryPolicy() {
    Map<Class<? extends Throwable>, Boolean> exceptionMap = new HashMap<>();

    exceptionMap.put(IllegalArgumentException.class, false);
    exceptionMap.put(TimeoutException.class, false);
    exceptionMap.put(ListenerExecutionFailedException.class, true);
    // Custom exception
    exceptionMap.put(MyTechnicalException.class, false);
    exceptionMap.put(MyFunctionalException.class, false);

    return new SimpleRetryPolicy(3, exceptionMap, true);
}

现在，如果我发送不可序列化的消息，我的消息将不重试地发送到DLT-

在我的@KafkaHandler中，我有一个抛出新的MyTechnicalException，捕获并重新捕获。

我应该没有重试，但我得到了2个重试，每个20秒（而不是10秒？），并在2次重试后向DLT发送了一条消息。

如果我删除errorHandler，不足为奇的是，我尝试了3次，并且显示了我的日志错误消息。。。但我需要发送给DLQ。。。

如果我删除RetryTemplate和RecoveryCallback，这并不奇怪，但所有异常都会重试。。。

问题：

1. 有办法处理我的用例吗？怎么做

Answer 1

由于错误处理程序已经发展到涵盖RetryTemplate提供的所有功能（回退、异常分类等），监听器级别的重试已被弃用。

https://github.com/spring-projects/spring-kafka/issues/1886