如何处理 Kafka Streams 中数据处理过程中发生的错误

Question

我正在使用 Spring Cloud Stream Kafka Streams 编写一个 Java 应用程序。这是我正在使用的函数方法 sn-p：

@Bean
public Function<KStream<String, String>, KStream<String, String>> process() {
    return input ->
        input.transform(
            () ->
                new Transformer<String, String, KeyValue<String, String>>() {

                  ProcessorContext context;

                  @Override
                  public void init(ProcessorContext context) {
                    this.context = context;
                  }

                  @Override
                  public void close() {}

                  @Override
                  public KeyValue<String, String> transform(String key, String value) {                       
                         String result = fetch_data_from_database(key, value);
                         return new KeyValue<>(key, result);
                  }
});

fetch_data_from_database() 可以抛出异常。

如果 fetch_from_database() 出现异常，我如何停止处理入站 KStream（偏移量不应被提交）并使用相同的偏移量数据重试处理？

Answer 1

在这种情况下，您需要自行重试逻辑。为此，您可以使用 Spring 的RetryTemplate。 This answer 详细介绍了如何在 Kafka Streams 中使用 RetryTemplate。它不像您那样使用低级处理器 API，但它是相同的想法。将您的数据库调用包装在重试模板中，并根据您的要求自定义重试。任何上游处理都将暂停，直到重试次数用尽。