错误处理
Apache Kafka Streams 提供了原生处理反序列化错误异常的能力。有关此支持的详细信息，请参阅此链接。开箱即用，Apache Kafka Streams 提供了两种反序列化异常处理器 - LogAndContinueExceptionHandler 和 LogAndFailExceptionHandler。顾名思义，前者会记录错误并继续处理下一条记录，而后者会记录错误并导致处理失败。LogAndFailExceptionHandler 是默认的反序列化异常处理器。
处理绑定器中的反序列化异常
Kafka Streams binder 允许使用以下属性来指定上述反序列化异常处理器。
spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: logAndContinue
或
spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: logAndFail
除了上述两种反序列化异常处理器外，binder 还提供了第三种处理器，用于将错误记录（毒丸）发送到 DLQ（死信队列）主题。以下是启用此 DLQ 异常处理器的方法。
spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: sendToDlq
当上述属性被设置后，所有反序列化错误的记录将自动发送到死信队列（DLQ）主题。
您可以将死信队列消息发布到的主题名称设置如下。
你可以为 DlqDestinationResolver 提供一个实现，它是一个函数式接口。DlqDestinationResolver 接收 ConsumerRecord 和异常作为输入，然后允许指定一个主题名称作为输出。通过访问 Kafka ConsumerRecord，可以在 BiFunction 的实现中检查头部记录。
以下是一个为 DlqDestinationResolver 提供实现的示例。
@Bean
public DlqDestinationResolver dlqDestinationResolver() {
return (rec, ex) -> {
if (rec.topic().equals("word1")) {
return "topic1-dlq";
}
else {
return "topic2-dlq";
}
};
}
在为 DlqDestinationResolver 提供实现时，需要牢记的一个重要事项是：绑定器中的配置器不会为应用程序自动创建主题。这是因为绑定器无法推断实现可能发送到的所有死信队列（DLQ）主题的名称。因此，如果你使用此策略提供 DLQ 名称，应用程序有责任确保这些主题已预先创建。
如果应用程序中存在名为 DlqDestinationResolver 的 bean，它将具有更高的优先级。如果您不希望采用这种方法，而是希望通过配置提供静态的 DLQ 名称，您可以设置以下属性。
spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.dlqName: custom-dlq (Change the binding name accordingly)
如果设置了此项，错误记录将被发送到主题 custom-dlq。如果应用程序未使用上述任一策略，则会创建一个名为 error.<input-topic-name>.<application-id> 的 DLQ 主题。例如，如果你的绑定目标主题是 inputTopic，应用程序 ID 是 process-applicationId，那么默认的 DLQ 主题就是 error.inputTopic.process-applicationId。如果打算启用 DLQ，始终建议为每个输入绑定显式创建一个 DLQ 主题。
每个输入消费者绑定的死信队列
属性 spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler 适用于整个应用程序。这意味着如果同一个应用程序中有多个函数，此属性将应用于所有函数。然而，如果您在单个处理器中有多个处理器或多个输入绑定，那么您可以使用绑定器为每个输入消费者绑定提供的更细粒度的 DLQ 控制。
如果你有以下处理器，
@Bean
public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, Long>> process() {
...
}
如果你只想在第一个输入绑定上启用 DLQ，并在第二个绑定上启用 skipAndContinue，那么你可以在消费者上按如下方式操作。
spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.deserializationExceptionHandler: sendToDlq spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.deserializationExceptionHandler: skipAndContinue
通过这种方式设置反序列化异常处理器的优先级高于在绑定器级别设置。
DLQ 分区
默认情况下，记录会被发布到死信主题，并使用与原记录相同的分区。这意味着死信主题的分区数量必须至少与原记录的分区数量相同。
要更改此行为，请在应用程序上下文中添加一个 DlqPartitionFunction 实现作为 @Bean。此类 bean 只能存在一个。该函数会接收消费者组（在大多数情况下与应用程序 ID 相同）、失败的 ConsumerRecord 和异常。例如，如果始终希望路由到分区 0，可以使用：
@Bean
public DlqPartitionFunction partitionFunction() {
return (group, record, ex) -> 0;
}
如果你将消费者绑定的 dlqPartitions 属性设置为 1（并且绑定器的 minPartitionCount 等于 1），则无需提供 DlqPartitionFunction；框架将始终使用分区 0。如果你将消费者绑定的 dlqPartitions 属性设置为大于 1 的值（或者绑定器的 minPartitionCount 大于 1），则必须提供一个 DlqPartitionFunction bean，即使分区数量与原始主题相同。
在使用 Kafka Streams 绑定器的异常处理功能时，需要注意以下几点。
-
属性
spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler适用于整个应用程序。这意味着如果同一个应用程序中有多个函数，此属性将应用于所有函数。 -
反序列化的异常处理与原生反序列化及框架提供的消息转换功能保持一致。
处理 Binder 中的生产异常
与上述反序列化异常处理程序的支持不同，绑定器并未为处理生产异常提供此类一流机制。然而，您仍然可以使用 StreamsBuilderFactoryBean 自定义器来配置生产异常处理程序，您可以在下面的后续章节中找到更多详细信息。
运行时错误处理
在处理来自应用程序代码（即业务逻辑执行）的错误时，通常由应用程序自行处理。因为Kafka Streams绑定器无法干预应用程序代码。然而，为了简化应用程序的处理流程，绑定器提供了一个便捷的RecordRecoverableProcessor，通过它，您可以指定如何处理应用程序级别的错误。
考虑以下代码。
@Bean
public java.util.function.Function<KStream<String, String>, KStream<String, String>> process() {
return input -> input
.map(...);
}
如果上述 map 调用中的业务代码抛出异常，您需要负责处理该错误。这正是 RecordRecoverableProcessor 发挥作用的地方。默认情况下，RecordRecoverableProcessor 只会记录错误并让应用程序继续运行。假设您希望将失败的记录发布到 DLT，而不是在应用程序内部处理。在这种情况下，您必须使用名为 DltAwareProcessor 的 RecordRecoverableProcessor 自定义实现。以下是具体操作方法。
@Bean
public java.util.function.Function<KStream<String, String>, KStream<String, String>> process(DltPublishingContext dltSenderContext) {
return input -> input
.process(() -> new DltAwareProcessor<>(record -> {
throw new RuntimeException("error");
}, "hello-dlt-1", dltPublishingContext));
}
原始 map 调用中的业务逻辑代码现已移至 KStream#process 方法调用中，该方法接收一个 ProcessorSupplier。接着，我们传入自定义的 DltAwareProcessor，该处理器能够发布到 DLT。上述 DltAwareProcessor 的构造函数接收三个参数：一个包含业务逻辑操作的 Function（其函数体处理输入记录）、DLT 主题，以及一个 DltPublishingContext。当 Function 的 lambda 表达式抛出异常时，DltAwareProcessor 会将输入记录发送到 DLT。DltPublishingContext 为 DltAwareProcessor 提供必要的发布基础设施 Bean。DltPublishingContext 由绑定器自动配置，因此您可以直接将其注入到应用程序中。
如果你不希望绑定器将失败的记录发布到 DLT，那么你必须直接使用 RecordRecoverableProcessor 而不是 DltAwareProcessor。你可以提供一个自定义的恢复器作为 BiConsumer，它接收输入 Record 和异常作为参数。假设一个场景，你不想将记录发送到 DLT，而只是记录消息并继续处理。下面是一个如何实现这一点的示例。
@Bean
public java.util.function.Function<KStream<String, String>, KStream<String, String>> process() {
return input -> input
.process(() -> new RecordRecoverableProcessor<>(record -> {
throw new RuntimeException("error");
},
(record, exception) -> {
// Handle the record
}));
}
在这种情况下，当记录处理失败时，RecordRecoverableProcessor会调用用户提供的恢复器，该恢复器是一个BiConsumer，它接收失败记录和抛出的异常作为参数。
在 DltAwareProcessor 中处理记录键
在使用 DltAwareProcessor 向 DLT 发送失败记录时，如果您希望将记录键发送到 DLT 主题，则需要在 DLT 绑定上设置正确的序列化器。这是因为 DltAwareProcessor 使用 StreamBridge，而 StreamBridge 使用常规的 Kafka 绑定器（基于消息通道），该绑定器默认对键使用 ByteArraySerializer。对于记录值，Spring Cloud Stream 会将有效负载转换为适当的 byte[]；但对于键则不然，因为它只是将标头中接收到的内容作为键传递。如果您提供的键不是字节数组，则可能会导致类转换异常，为避免这种情况，您需要在 DLT 绑定上设置序列化器，如下所示。
假设DLT目标为hello-dlt-1，且记录键的数据类型为String。
spring.cloud.stream.kafka.bindings.hello-dlt-1.producer.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer