处理Kafka流中的异常

这取决于你想做什么,除了生产者方面的例外。 如果将对生产者抛出异常(例如,由于网络故障或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 如果你想要流停止(失败是默认的)。 您需要在流配置中指定该类:

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

还要注意 ProductionException处理程序 只处理Producer上的异常,并且在使用流方法处理消息时不处理异常 mapValues(..) 我是说, filter(..) 我是说, branch(..) 等等,您需要用try/catch块包装这些方法逻辑(将所有方法逻辑放入try块中,以确保您将处理所有异常情况):

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

我知道,我们不需要显式地处理消费者端的异常,因为卡夫卡流稍后会自动重试消费(因为在消费和处理消息之前,偏移量不会更改);例如,如果卡夫卡代理在一段时间内不可访问,您将从卡夫卡流获得异常,当中断时将被中断,卡夫卡流将消耗所有消息。所以在这种情况下,我们只会有延迟,没有任何损坏/丢失。

具有 setUncaughtExceptionHandler 您将无法更改默认行为,如 ProductionException处理程序 ,使用它,您只能记录错误或将消息发送到失败主题。