处理Kafka流中的异常

Question

曾发表过多篇文章，但大多数都与处理错误消息有关，而不是处理过程中的异常处理。

我想知道如何处理流应用程序接收到的消息，并且在处理消息时出现异常？异常可能是由于多种原因造成的，如网络故障、RuntimeException等。，

• 有人能提出正确的方法吗？我应该使用setUncaughtExceptionHandler吗？还是有更好的方法

Answer 1

setUncaughtExceptionHandler不帮助处理异常，它在流因未捕获的异常而终止后工作。

Kafka提供了几种处理异常的方法。简单的try-catch{}有助于捕获处理器代码中的异常，但kafka反序列化异常（可能是由于数据问题）和生产异常（在与代理通信期间发生）分别需要反序列化异常处理程序和生产异常处理程序。默认情况下，如果Kafka应用程序遇到以下任何一种情况，它都会失败。

你可以在这个帖子上找到

Answer 2

用于处理消费者方面的异常，

1） 您可以使用以下属性在producer中添加默认异常处理程序。

"default.deserialization.exception.handler" = "org.apache.kafka.streams.errors.LogAndContinueExceptionHandler";

基本上，apache提供了三个异常处理程序类

1） LogAndContiuneExceptionHandler，您可以将其作为

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, 
           LogAndContinueExceptionHandler.class);

2） LogAndFailExceptionHandler

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, 
           LogAndFailExceptionHandler.class);

3） LogAndSkipOnInvalidTimestamp

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, 
           LogAndSkipOnInvalidTimestamp.class);

对于自定义异常处理，

1）您可以实现DeserializationExceptionHandler接口并覆盖句柄（）方法。

2） 也可以扩展上述类。

Answer 3

这取决于你想做什么，除了生产者方面的例外。若将在生产者上引发异常（例如，由于网络故障或kafka broker已死亡），则默认情况下，流将死亡。使用kafka streams 1.1.0版，您可以通过实现ProductionExceptionHandler来覆盖默认行为，如下所示：

public class CustomProductionExceptionHandler implements ProductionExceptionHandler {

    @Override
    public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
                                                     final Exception exception) {
        log.error("Kafka message marked as processed although it failed. Message: [{}], destination topic: [{}]",  new String(record.value()), record.topic(), exception);
        return ProductionExceptionHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(final Map<String, ?> configs) {
    }

}

从handle方法，如果不希望流在异常时消亡，可以返回CONTINUE，如果希望流停止，可以返回FAIL（默认为FAIL）。您需要在流配置中指定此类：

default.production.exception.handler=com.example.CustomProductionExceptionHandler

另请注意，ProductionExceptionHandler只处理生产者上的异常，它不会在使用流方法mapValue（...）、filter（...）、分支（...）等处理消息期间处理异常，您需要将这些方法逻辑包装为try/catch块（将您的所有方法逻辑放入try块以保证您将处理所有异常情况）：

.filter((key, value) -> { try {..} catch (Exception e) {..} })

正如我所知，我们不需要显式地处理消费者端的异常，因为kafka流稍后会自动重试消费（因为在消费和处理消息之前，偏移量不会更改）；e、 g.如果Kafka代理在一段时间内无法访问，您将从Kafka流中获得异常，并且当中断时，Kafka流将消耗所有消息。因此，在这种情况下，我们只能延迟，没有任何损坏/丢失。

使用setUncaughtExceptionHandler您将无法更改默认行为，例如使用ProductionExceptionHandler，使用它您只能将错误记录或发送消息到失败主题。

自kafka streams 2.8.0以来更新

由于kafka streams（Kafka流）是2.8.0版，因此您可以使用KafkaStreams（Kafka流）方法自动替换失败的流线程（由未捕获的异常引起

kafkaStreams.setUncaughtExceptionHandler(ex -> {
    log.error("Kafka-Streams uncaught exception occurred. Stream will be replaced with new thread", ex);
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
});