From 090acc408f7fb17aede0c0d7dff705be8124df71 Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Thu, 3 Mar 2022 14:36:05 -0500
Subject: [PATCH] Kafka Streams binder uncaught exception changes

The default StreamsUncaughtExceptionHandler is added only if the
application does not provide one.
---
 pom.xml                                                     | 2 +-
 .../binder/kafka/streams/StreamsBuilderFactoryManager.java  | 6 ++++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/pom.xml b/pom.xml
index 87c5d5029..da4ff50e9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -21,7 +21,7 @@
 	</scm>
 	<properties>
 		<java.version>1.8</java.version>
-		<spring-kafka.version>2.8.2</spring-kafka.version>
+		<spring-kafka.version>2.8.4-SNAPSHOT</spring-kafka.version>
 		<spring-integration-kafka.version>5.5.8</spring-integration-kafka.version>
 		<kafka.version>3.0.0</kafka.version>
 		<spring-cloud-stream.version>3.2.3-SNAPSHOT</spring-cloud-stream.version>
diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
index 9f2bbf0f6..b8c0277df 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
@@ -96,8 +96,10 @@ public class StreamsBuilderFactoryManager implements SmartLifecycle {
 					// By default, we shut down the client if there is an uncaught exception in the application.
 					// Users can override this by customizing SBFB. See this issue for more details:
 					// https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1110
-					streamsBuilderFactoryBean.setStreamsUncaughtExceptionHandler(exception ->
-							StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
+					if (streamsBuilderFactoryBean.getStreamsUncaughtExceptionHandler() == null) {
+						streamsBuilderFactoryBean.setStreamsUncaughtExceptionHandler(exception ->
+								StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
+					}
 					// Starting the stream.
 					final Map<StreamsBuilderFactoryBean, List<ConsumerProperties>> bindingServicePropertiesPerSbfb =
 							this.kafkaStreamsBindingInformationCatalogue.getConsumerPropertiesPerSbfb();