Spring Kafka中KafkaStreams是否支持非阻塞重试？

• Spring Kafka的非阻塞重试（Retry Topic）功能目前仅针对@KafkaListener注解的消费者，并没有为Kafka Streams提供开箱即用的同类型非阻塞重试支持。

• 不过可以结合Kafka Streams自身的错误处理机制，搭配Spring Kafka的配置实现类似的非阻塞重试效果，常见实现方案如下：

  1. 死信主题（DLT）+ 自定义重试转发逻辑
     通过配置StreamsBuilderFactoryBean的默认生产异常处理器，将处理失败的消息转发到重试主题，再通过独立的消费组件（Streams拓扑或@KafkaListener）消费重试主题完成重试，示例代码：

     @Bean
     public StreamsBuilderFactoryBeanCustomizer streamsCustomizer() {
         return factoryBean -> {
             factoryBean.setDefaultProductionExceptionHandler(new DefaultProductionExceptionHandler() {
                 @Override
                 public ProductionExceptionHandlerResponse handle(ProducerRecord<byte[], byte[]> record, Exception exception, Producer<byte[], byte[]> producer) {
                     // 将失败消息发送到指定重试主题
                     producer.send(new ProducerRecord<>("my-topic-retry", record.key(), record.value()));
                     return ProductionExceptionHandlerResponse.CONTINUE;
                 }
             });
         };
     }
     

  2. 自定义Processor实现重试控制
     在Streams拓扑中添加自定义处理器，捕获业务处理异常后将消息转发到重试主题，同时可以结合延迟队列或定时器控制重试间隔，示例代码：

     @Bean
     public KStream<String, String> kStream(StreamsBuilder builder) {
         KStream<String, String> inputStream = builder.stream("input-topic");
         
         inputStream.process(() -> new AbstractProcessor<String, String>() {
             @Override
             public void process(String key, String value) {
                 try {
                     // 执行业务处理逻辑
                     handleBusinessLogic(value);
                 } catch (Exception e) {
                     // 转发到重试主题
                     context().forward(key, value, To.child("retry-sink"));
                 }
             }
         });
         
         // 绑定重试主题输出
         inputStream.to("retry-topic", Produced.with(Serdes.String(), Serdes.String()));
         return inputStream;
     }
     

• 若需要指数退避、重试次数限制等精细控制，可以结合Spring Retry框架，但需手动管理重试主题的创建、重试间隔配置等逻辑，因为Spring Kafka的Retry Topic自动管理功能暂不支持Kafka Streams。

内容的提问来源于stack exchange，提问作者Varun Arora