exception - 拓扑发生异常时跳过记录

问题描述

我们正在编写一个 Kafka Streams Topology 来聚合数据并实时显示它们。我们希望使显示尽可能健壮 - 理想情况下记录记录并继续出现任何异常。

根据文档，我们的一些测试和

Kafka Streams 很好地支持处理 Producer 或反序列化期间发生的异常。提供LogAndContinueExceptionHandler的正是我们想要的行为。然而我们的主要问题是处理过程中发生的异常（例如在.mapValues()或.leftJoin()

我们的想法基本上是验证先决条件

在反序列化过程中，如果未完成，则抛出 DeserializationException（并记录并继续）。
如果无法执行计算（/ by zero error等），则在处理函数中检查以返回默认值

但是，如果数据中有不可预见的东西，异常仍然可能冒出，并且拓扑将关闭。

Kafka Streams 提供了一个，UncaughtExceptionHandler但它是在线程已经死亡后调用的，因此它不能用于防止拓扑关闭。

有没有办法编写一个跳过记录的 UncaughtExceptionHandler ？或者另一种机制来跳过我们可以在try-catch处理函数内部的块中的当前记录？

标签： exceptionapache-kafkaapache-kafka-streams

解决方案

我认为最好的解决方案是以永远不会抛出任何异常的方式编写处理操作（例如：映射器、过滤器等）。为此，您可以使用一个包装器对象，它可以是成功或错误（例如：Eitherscala 中的类型）。之后，您可以使用该branch()方法获取两个流：一个用于成功记录，一个用于错误记录。

下面的代码显示了基本思想：

    public static void main(String[] args) {
        var builder = new StreamsBuilder();
        KStream<Object, Result<Object>> stream = builder.stream("my-topic")
            .map((k, v) -> {
                try {
                    // unsafe operation, i.e that may throw an exception
                    return KeyValue.pair(k, new Success<>(v));
                } catch (Exception e) {
                    return KeyValue.pair(k, new Error<>(e));
                }
            });
        KStream<Object, Result<Object>>[] branch = stream.branch((k, v) -> !v.hasError(), (k, v) -> v.hasError());

        // Handle the success steam
        KStream<Object, Result<Object>> successStream = branch[0];

        // Handle the error steam, e.g:  log errors, write errors to a Dead Letter Queue
        KStream<Object, Result<Object>> errorStream = branch[1];
        
    }

    public interface Result<T> {
        T get() throws Exception;
        Exception exception();
        boolean hasError();
    }

    public static class Success<T> implements Result<T> {

        private final T value;

        public Success(T value) {
            this.value = value;
        }

        @Override
        public T get() throws Exception {
            return value;
        }

        @Override
        public Exception exception() {
            return null;
        }

        @Override
        public boolean hasError() {
            return false;
        }
    }

    public static class Error<T> implements Result<T> {

        private final Exception error;

        public Error(Exception error) {  this.error = error; }

        @Override
        public T get() throws Exception{
            throw error;
        }

        @Override
        public Exception exception() {
            return error;
        }

        @Override
        public boolean hasError() {
            return true;
        }
    }

另外，对于您提到的反序列化异常，Azkarra Streams项目提供了一些方便的 java 类，可以帮助您（例如 SafeSerdes、DeadLetterTopicExceptionHandler）：GitHub

exception - 拓扑发生异常时跳过记录

问题描述

解决方案

推荐阅读