Spring云Kafka Stream StreamsUncaughtExceptionHandler

Question

我正在尝试将StreamsUncaughtExceptionHandler添加到我的Kafka流处理器。该处理器使用Kafka函数编写。我查看了Artem Bilan提供的将StreamsUncaughtExceptionHandler包括到我的服务中的建议，但我的异常从未被它捕获/处理。

配置Bean：

@Autowired
UnCaughtExceptionHandler exceptionHandler;

@Bean
public StreamsBuilderFactoryBeanConfigurer streamsCustomizer() {
    return new StreamsBuilderFactoryBeanConfigurer() {

        @Override
        public void configure(StreamsBuilderFactoryBean factoryBean) {;
            factoryBean.setStreamsUncaughtExceptionHandler(exceptionHandler);
        }

        @Override
        public int getOrder() {
            return Integer.MAX_VALUE;
        }

    };
}

自定义异常处理程序：

    @Component
public class UnCaughtExceptionHandler implements StreamsUncaughtExceptionHandler {

  @Autowired
  private StreamBridge streamBridge;

  @Override
  public StreamThreadExceptionResponse handle(Throwable exception) {
    return StreamThreadExceptionResponse.REPLACE_THREAD;
  }
}

流处理功能：

@Autowired
private MyService service;

@Bean
public Function<KStream<String, Input>, KStream<String, Output>> processor() {
    final AtomicReference<KeyValue<String, Output>> result = new AtomicReference<>(null);
    return kStream -> kStream
            .filter((key, value) -> value != null)
            .filter((key, value) -> {
                Optional<Output> outputResult = service.process(value);
                if (outputResult.isPresent()) {
                    result.set(new KeyValue<>(key, outputResult.get()));
                    return true;
                }
                return false;
            })
        .map((messageKey, messageValue) -> result.get());
}

我希望UnCaughtExceptionHandler处理service.process（）方法引发的任何异常。但是异常永远不会进入句柄方法；相反，它们会传播到根并终止客户端。我也看了这个解决方案，但我想以更独立的方式处理它。

问题：如何使用StreamsUncaughtExceptionHandler处理任何处理异常？

• Spring启动版本：2.6.3
• Spring-Cloud-stream版本：3.2.1
• 春-云-流-绑定器-Kafka-流：3.2.1
• Kafka流：3.0.0

可复制的示例：spring cloud kafka streams异常

Answer 1

以下是一些你可以尝试的事情。

>

我注意到您为配置的impl中的顺序设置了intger.MAX_VALUE。默认情况下，StreamsBuilderFactoryBean使用的相位值为intger.MAX_VALUE-1000，因此在工厂bean准备好启动时，配置程序可能还不可用，因为intger.MAX_VALUE的优先级较低。您可以将订单更改为intger.MAX_VALUE-5000，以确保在工厂bean启动之前配置bean已完全实例化。

从这些选项开始，看看它们是否给出了该问题的任何指示。如果它仍然存在，请随时与我们分享一个可重复的小样本应用程序。