如何在 Spring Kafka Stream 中配置 UncaughtExceptionHandler

Question

我正在使用 spring-kafka 来实现一个使用 Spring Boot 1.5.16 的流应用程序。我们使用的spring-kafka 版本是1.3.8.RELEASE。

我正在寻找一种方法来关闭启动应用程序，以防出现终止与 Kafka Streams 关联的所有线程的错误。我发现在KafkaStreams 中可以注册未捕获异常的句柄。方法是setGlobalStateRestoreListener。

我看到这个方法暴露在spring-kafka 中，类型为KStreamBuilderFactoryBean。

我的问题如下。有没有一种简单的方法可以将UncaughtExceptionHandler 注册为 bean 并让 Spring 在工厂 bean 中正确注入？还是我应该自己创建KStreamBuilderFactoryBean 并手动设置处理程序？

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME)
public KStreamBuilderFactoryBean kStreamBuilderFactoryBean(StreamsConfig streamsConfig) {
    final KStreamBuilderFactoryBean streamBuilderFactoryBean = new KStreamBuilderFactoryBean(
            streamsConfig);
    streamBuilderFactoryBean.setUncaughtExceptionHandler((threadInError, exception) -> {
        // Something happens here
    });
    return streamBuilderFactoryBean;
}

非常感谢。

Answer 1

是的。在那个旧版本中，您必须自己指定一个 KStreamBuilderFactoryBean bean，并进行适当的注入，而正是 KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME。

在以后的版本中，我们已经有了一个 StreamsBuilderFactoryBeanConfigurer 来保持自动配置的 KStreamBuilderFactoryBean，但可以根据需要进行任何修改。

更新

您只需在应用程序上下文中将其创建为 bean，框架就会将其拾取并应用于 StreamsBuilderFactoryBean：

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

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

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

        };
    }