Spring Cloud Kafka流中的错误处理

Question

我用Spring云溪和Kafka溪。假设我有一个处理器，它的功能是将KStream字符串转换为KStream CityProgrammes。它调用一个API来根据名称查找城市，并调用另一个转换来查找该城市附近的任何事件。

现在的问题是，任何错误发生在转换期间，整个应用程序停止。我想把一个特定的消息发送给DLQ，然后继续前进。我已经读了几天了，每个人都建议在被调用的服务中处理错误，但在我看来这是一个不必要的事情，而且我仍然需要返回一个kstream：我如何在catch中做到这一点？

我还查看了UncaughtExeptionHandler，但它不知道消息，只能重新启动处理，不会跳过这个无效的消息。

这可能听起来像是一个A-B问题，所以问题的措辞是：当发生异常并将无效项发送到DLQ时，我如何维护KStream中的流？

Answer 1

当涉及到应用程序级别的错误时，如何处理错误取决于应用程序本身。Kafka Streams和Spring Cloud Stream绑定器主要支持框架级别的反序列化和序列化错误。虽然是这样，但我认为你的场景是可以处理的。如果您使用的是Kafka客户端2.8之前的版本，这里有一个类似的答案:https://stackoverflow.com/A/66749750/2070861

如果您正在使用Kafka/Streams2.8，这里有一个您可以使用的想法。但是，下面的代码应该只作为一个起点。根据您的用例进行调整。阅读更多关于Kafka Streams2.8中分支如何工作的信息。分支API在2.8中与以前的版本进行了显著的重构。

public Function<KStream<?, String>, KStream<?, Foo>> convert() {
            Foo[] foo = new Foo[0];
            return input -> {
                final Map<String, ? extends KStream<?, String>> branches =
                        input.split(Named.as("foo-")).branch((key, value) -> {
                                    try {
                                        foo[0] = new Foo(); // your API call for CitiProgramme converion here, possibly.
                                        return true;
                                    }
                                    catch (Exception e) {
                                        Message<?> message = MessageBuilder.withPayload(value).build();
                                        streamBridge.send("to-my-dlt", message);
                                        return false;
                                    }

                                }, Branched.as("bar"))
                                .defaultBranch();

                final KStream<?, String> kStream = branches.get("foo-bar");
                return kStream.map((key, value) -> new KeyValue<>("", foo[0]));
            };

        }


    }

默认分支在此代码中被忽略，因为它只包含引发异常的记录。这些由上面的catch语句处理，在该语句中，我们以编程方式将记录发送到DLT。最后，我们获取好的记录并将它们映射到新的kstream并通过出站发送。