如何在Spring Kafka Stream中配置UncaughtExceptionHandler

Question

我正在使用spring kafka来实现一个使用spring Boot 1.5.16的流应用程序。我们使用的SpringKafka版本是1.3.8。释放。

我正在搜索一种方法，以便在出现终止与Kafka流关联的所有线程的错误时关闭启动应用程序。我发现在KafkaStreams中，有可能为未捕获的异常注册句柄。方法是setGlobalStateRestoreListener。

我看到这个方法在类型kstreambilderfactorybean中的SpringKafka中公开了。

我的问题如下。有没有一种简单的方法可以将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

对在旧版本中，您必须通过适当的注入和KafkaStreamsDefaultConfiguration自己指定一个KStreamBuilderFactoryBean。默认\u KSTREAM\u BUILDER\u BEAN\u 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;
            }

        };
    }