SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1110: Kafka Streams state machine changes

Browse Source 
Restore Kafka Streams error state behavior in the binder, equivalent
to Kafka Streams prior to 2.8. Starting with 2.8, users can customize
the way uncaught errors are interpreted via an UncaughtExceptionHandler
in the applicaiton. Binder now sets a default handler that shuts down
the Kafka Streams client if thre is an uncaught error.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1110
This commit is contained in:
Soby Chacko
2021-07-29 12:17:59 -04:00
committed by Gary Russell
parent 912c47e3ac
commit a5f01f9d6f
2 changed files with 8 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
View File

@@ -18,6 +18,8 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Set;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.kafka.KafkaException;
@@ -84,6 +86,12 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
if (this.listener != null) {
streamsBuilderFactoryBean.addListener(this.listener);
}
// By default, we shutdown 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);
// Starting the stream.
streamsBuilderFactoryBean.start();
this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
}

13
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java
View File

@@ -257,19 +257,6 @@ public class KafkaStreamsBinderHealthIndicatorTests {
});
}
@Bean
public StreamsBuilderFactoryBeanConfigurer customizer() {
return factoryBean -> {
factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
@Override
public void customize(KafkaStreams kafkaStreams) {
kafkaStreams.setUncaughtExceptionHandler(exception ->
StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
}
});
};
}
}
@EnableBinding({ KafkaStreamsProcessor.class, KafkaStreamsProcessorX.class })