此版本仍在开发中,尚不被认为是稳定的。对于最新的稳定版本,请使用 Spring for Apache Kafka 3.3.9! |
DLT 策略
该框架提供了一些使用 DLT 的策略。 可以提供 DLT 处理方法,使用默认日志记录方法,或者根本没有 DLT。 此外,您还可以选择如果 DLT 处理失败会发生什么。
DLT处理方法
您可以指定用于处理主题的 DLT 的方法,以及该处理失败时的行为。
为此,您可以使用@DltHandler
类的方法中的注释,其中包含@RetryableTopic
注释。
请注意,相同的方法将用于所有@RetryableTopic
该类中的 Commentted 方法。
@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 处理失败,则有两种可能的行为可用: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 . |
从 2.8.3 版本开始,ALWAYS_RETRY_ON_ERROR 如果记录导致引发致命异常,则不会将记录路由回 DLT,
例如DeserializationException ,因为通常总是会抛出此类异常。 |
被视为致命的例外情况是:
-
DeserializationException
-
MessageConversionException
-
ConversionException
-
MethodArgumentResolutionException
-
NoSuchMethodException
-
ClassCastException
您可以使用DestinationTopicResolver
豆。
有关详细信息,请参阅异常分类器。
配置无 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);
}