Repository: quarkusio/quarkus
Issue: KafkaStreams UncaughtExceptionHandler can only be set before starting the stream

## Describe the bug
It is not possible to use KafkaStreams#setUncaughtExceptionHandler on the injected `KafkaStreams` instance because it is already started by the `KafkaStreamsProducer` managed by kafka-streams Quarkus extension.

### Expected behavior
The kafka-streams extension calls setUncaughtExceptionHandler with an instance of `StreamsUncaughtExceptionHandler` discovered in the application the before the start, much like StateListener and StateRestoreListener

Reported by @andrerigon


Changed files:
extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java

diff --git a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
index 408fa5d0957..93ca55cbb33 100644
--- a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
+++ b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
@@ -39,6 +39,7 @@
 import org.apache.kafka.streams.KafkaStreams.StateListener;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
 import org.apache.kafka.streams.processor.StateRestoreListener;
 import org.eclipse.microprofile.config.ConfigProvider;
 import org.jboss.logging.Logger;
@@ -73,7 +74,8 @@ public class KafkaStreamsProducer {
     public KafkaStreamsProducer(KafkaStreamsSupport kafkaStreamsSupport, KafkaStreamsRuntimeConfig runtimeConfig,
             Instance<Topology> topology, Instance<KafkaClientSupplier> kafkaClientSupplier,
             @Identifier("default-kafka-broker") Instance<Map<String, Object>> defaultConfiguration,
-            Instance<StateListener> stateListener, Instance<StateRestoreListener> globalStateRestoreListener) {
+            Instance<StateListener> stateListener, Instance<StateRestoreListener> globalStateRestoreListener,
+            Instance<StreamsUncaughtExceptionHandler> uncaughtExceptionHandlerListener) {
         shutdown = false;
         // No producer for Topology -> nothing to do
         if (topology.isUnsatisfied()) {
@@ -104,7 +106,8 @@ public KafkaStreamsProducer(KafkaStreamsSupport kafkaStreamsSupport, KafkaStream
         this.executorService = Executors.newSingleThreadExecutor();
 
         this.kafkaStreams = initializeKafkaStreams(kafkaStreamsProperties, runtimeConfig, kafkaAdminClient, topology.get(),
-                kafkaClientSupplier, stateListener, globalStateRestoreListener, executorService);
+                kafkaClientSupplier, stateListener, globalStateRestoreListener, uncaughtExceptionHandlerListener,
+                executorService);
         this.kafkaStreamsTopologyManager = new KafkaStreamsTopologyManager(kafkaAdminClient);
     }
 
@@ -149,7 +152,7 @@ private static KafkaStreams initializeKafkaStreams(Properties kafkaStreamsProper
             KafkaStreamsRuntimeConfig runtimeConfig, Admin adminClient, Topology topology,
             Instance<KafkaClientSupplier> kafkaClientSupplier,
             Instance<StateListener> stateListener, Instance<StateRestoreListener> globalStateRestoreListener,
-            ExecutorService executorService) {
+            Instance<StreamsUncaughtExceptionHandler> uncaughtExceptionHandlerListener, ExecutorService executorService) {
         KafkaStreams kafkaStreams;
         if (kafkaClientSupplier.isUnsatisfied()) {
             kafkaStreams = new KafkaStreams(topology, kafkaStreamsProperties);
@@ -163,6 +166,9 @@ private static KafkaStreams initializeKafkaStreams(Properties kafkaStreamsProper
         if (!globalStateRestoreListener.isUnsatisfied()) {
             kafkaStreams.setGlobalStateRestoreListener(globalStateRestoreListener.get());
         }
+        if (!uncaughtExceptionHandlerListener.isUnsatisfied()) {
+            kafkaStreams.setUncaughtExceptionHandler(uncaughtExceptionHandlerListener.get());
+        }
 
         executorService.execute(new Runnable() {