Spring cloud Kafka Stream StreamsUncaughtExceptionHandler

Question

I am trying to add StreamsUncaughtExceptionHandler to my Kafka stream processor. This processor is written with Kafka functions. I had a look at the suggestion provided by Artem Bilan to include the StreamsUncaughtExceptionHandler to my service, but my exceptions never get caught/handled by it.

Config 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;
        }

    };
}

Customized exception handler:

    @Component
public class UnCaughtExceptionHandler implements StreamsUncaughtExceptionHandler {

  @Autowired
  private StreamBridge streamBridge;

  @Override
  public StreamThreadExceptionResponse handle(Throwable exception) {
    log.error("Exception ", exception);
    // TODO handle exception
    return StreamThreadExceptionResponse.REPLACE_THREAD;
  }
}

Stream Process function:

@Autowired
private InternalService service;

@Bean
@SuppressWarnings("unchecked")
public Function<KStream<String, Internal>, KStream<String, Output>> showProcessor() {
    final AtomicReference<KeyValue<String, Output>> result = new AtomicReference<>(null);
    return kStream -> kStream
            .filter((key, value) -> value != null)
            .peek((key, value) -> print(value))
            .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());
}

I expect UnCaughtExceptionHandler to handle any exceptions thrown by the service.process() method. But exceptions never get into the handle method; instead, they propagate to the root and die the client. Had a look at this solution as well, but I want to handle it more independent way.

Question: How do I handle any processing exceptions using StreamsUncaughtExceptionHandler?

• Spring boot version: 2.6.3
• spring-cloud-stream version: 3.2.1
• spring-cloud-stream-binder-kafka-streams: 3.2.1
• kafka-streams: 3.0.0