错误处理

Apache Kafka Streams 提供了原生处理反序列化错误的异常处理能力。有关此支持的详细信息，请参阅 此文。开箱即用，Apache Kafka Streams 提供了两种反序列化异常处理器 - LogAndContinueExceptionHandler 和 LogAndFailExceptionHandler。顾名思义，前者将记录错误并继续处理下一个记录，而后者将记录错误并失败。LogAndFailExceptionHandler 是默认的反序列化异常处理器。

在绑定器中处理反序列化异常

Kafka Streams 绑定器允许使用以下属性指定上述反序列化异常处理器。

spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: logAndContinue

或

spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: logAndFail

除了上述两种反序列化异常处理器，绑定器还提供了第三种，用于将错误记录（毒丸）发送到 DLQ（死信队列）主题。以下是启用此 DLQ 异常处理器的方法。

spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler: sendToDlq

当设置上述属性时，所有反序列化错误的记录将自动发送到 DLQ 主题。

您可以设置 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 的实现时，需要记住一件重要的事情，那就是绑定器中的 provisioner 不会自动为应用程序创建主题。这是因为绑定器无法推断出实现可能发送到的所有 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 主题。

每个输入消费者绑定一个 DLQ

属性 spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler 适用于整个应用程序。这意味着，如果同一个应用程序中有多个函数，此属性将应用于所有这些函数。但是，如果单个处理器中有多个处理器或多个输入绑定，则可以使用绑定器为每个输入消费者绑定提供的更细粒度的 DLQ 控制。

如果您有以下处理器，

@Bean
public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, Long>> process() {
...
}

并且您只想在第一个输入绑定上启用 DLQ，并在第二个绑定上跳过并继续，那么您可以在消费者上按如下方式进行操作。

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 适用于整个应用程序。这意味着，如果同一个应用程序中有多个函数，此属性将应用于所有这些函数。

  • 反序列化的异常处理与原生反序列化和框架提供的消息转换保持一致。

在绑定器中处理生产异常

与上述反序列化异常处理器的支持不同，绑定器不提供此类一流的机制来处理生产异常。但是，您仍然可以使用 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，而不是在应用程序内部处理它。在这种情况下，您必须使用 RecordRecoverableProcessor 的自定义实现，称为 DltAwareProcessor。以下是实现方法。

@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，它将输入记录作为输入，然后将业务逻辑操作作为 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，它使用常规的 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
© . This site is unofficial and not affiliated with VMware.