DLT 策略

Doubao 中英对照 DLT Strategies

框架提供了几种处理 DLT 的策略。你可以提供一个用于 DLT 处理的方法，使用默认的日志方法，或者完全不使用 DLT。此外，你还可以选择 DLT 处理失败时的行为。

DLT 处理方法

你可以指定用于处理该主题 DLT 的方法，以及该处理失败时的行为。

为此，你可以在带有 @RetryableTopic 注解的类的方法上使用 @DltHandler 注解。请注意，该类中所有带 @RetryableTopic 注解的方法都将使用同一个方法。

@RetryableTopic
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

@DltHandler
public void processDltMessage(MyPojo message) {
    // ... message processing, persistence, etc
}

DLT 处理器方法也可以通过 RetryTopicConfigurationBuilder.dltHandlerMethod(String, String) 方法提供，参数为应处理 DLT 消息的 bean 名称和方法名。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltHandlerMethod("myCustomDltProcessor", "processDltMessage")
            .create(template);
}

@Component
public class MyCustomDltProcessor {

    private final MyDependency myDependency;

    public MyCustomDltProcessor(MyDependency myDependency) {
        this.myDependency = myDependency;
    }

    public void processDltMessage(MyPojo message) {
        // ... message processing, persistence, etc
    }
}

备注

如果未提供 DLT 处理器，则使用默认的 RetryTopicConfigurer.LoggingDltListenerHandlerMethod。

从 2.8 版本开始，如果你根本不想在此应用程序中消费 DLT（包括默认处理器的消费）（或者你希望延迟消费），你可以控制 DLT 容器是否启动，这与容器工厂的 autoStartup 属性无关。

使用 @RetryableTopic 注解时，将 autoStartDltHandler 属性设置为 false；使用配置构建器时，使用 autoStartDltHandler(false)。

之后你可以通过 KafkaListenerEndpointRegistry 启动 DLT 处理器。

DLT 失败行为

如果 DLT 处理失败，有两种可用行为：ALWAYS_RETRY_ON_ERROR 和 FAIL_ON_ERROR。

前者会将记录转发回 DLT 主题，使其不会阻塞其他 DLT 记录的处理；后者则消费者直接结束执行，不转发消息。

@RetryableTopic(dltProcessingFailureStrategy =
            DltStrategy.FAIL_ON_ERROR)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltHandlerMethod("myCustomDltProcessor", "processDltMessage")
            .doNotRetryOnDltFailure()
            .create(template);
}

备注

默认行为是 ALWAYS_RETRY_ON_ERROR。

important

从 2.8.3 版本开始，如果一条记录导致抛出致命异常（例如 DeserializationException），ALWAYS_RETRY_ON_ERROR 不会将该记录路由回 DLT，因为这类异常通常总是会被抛出。

被视为致命的异常有：

• DeserializationException
• MessageConversionException
• ConversionException
• MethodArgumentResolutionException
• NoSuchMethodException
• ClassCastException

你可以使用 DestinationTopicResolver bean 上的方法向此列表添加异常或从中移除异常。

更多信息请参阅异常分类器。

配置不使用 DLT

框架也允许不为该主题配置 DLT。在这种情况下，重试次数耗尽后处理就直接结束。

@RetryableTopic(dltProcessingFailureStrategy =
            DltStrategy.NO_DLT)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .doNotConfigureDlt()
            .create(template);
}