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

Question

我正在使用Spring云流Kafka流编写Java应用程序。下面是我正在使用的函数方法片段：

@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\u from\u database（）发生异常，如何停止对入站KStream的处理（不应提交偏移量），并使其使用相同的偏移量数据重试处理？

Answer 1

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