如何停止发送到Kafka主题当控件去抓块函数Kafkaspring

Question

你能告诉我，我如何停止发送到我的第三个Kafka主题，当控件到达catch块时，当前消息被发送到错误主题以及在正常处理情况下它应该发送到的主题。一段代码如下所示：

@Component
public class Abc {
private final StreamBridge streamBridge;
public Abc (StreamBridge streamBridge)
this.streamBridge = streamBridge;
@Bean
public Function<KStream<String, KafkaClass>, KStream<String,KafkaClass>> hiProcess() {
return input -> input.map((key,value) -> {
try{
KafkaClass stream = processFunction();
}
catch(Exception e) {
Message<KakfaClass> mess = MessageBuilder.withPayload(value).build();
streamBridge.send("errProcess-out-0". mess);
}
return new KeyValue<>(key, stream);
})
}
}

    

Answer 1

这可以使用以下模式实现：

KafkaClass stream;
return input -> input
    .branch((k, v) -> {
        try {
            stream = processFunction();
            return true;
        }
        catch (Exception e) {
            Message<KakfaClass> mess = MessageBuilder.withPayload(value).build();
            streamBridge.send("errProcess-out-0". mess);
            return false;
        }
       },
       (k, v) -> true)[0]
    .map((k, v) -> new KeyValue<>(k, stream));

这里，我们使用kstream的分支特性(API)将输入分为两个路径--正常流和导致错误的流。这是通过向branch方法调用提供两个筛选器来实现的。第一个过滤器是调用processfunction方法并返回响应的正常流。如果我们没有得到异常，过滤器返回true，分支操作的结果在输出数组的第一个元素[0]中捕获，该元素在map操作的下游处理，并将最终结果发送给出站主题。

另一方面，如果抛出异常，则使用StreamBridge向错误主题发送所需的内容，过滤器返回false。由于下游map操作只对分支[0]的数组的第一个元素执行，因此不会向外发送任何内容。当第一个筛选器返回false时，它会转到第二个筛选器，后者总是返回true。这是一个完全忽略结果的无操作筛选器。

这种特殊实现的一个缺点是，您需要将来自processfunction的响应存储在实例字段中，然后对每个传入的kstream记录进行变异，以便在发送输出的最终map方法中访问其值。然而，对于这个特定的用例，这可能不是一个问题。