From becdfefbedb4a02a7a83ed54e0f74b3e5262e48f Mon Sep 17 00:00:00 2001 From: Trygve Aasjord Date: Thu, 14 May 2026 08:43:48 +0200 Subject: [PATCH 1/2] Add Kafka Stream's leaveGroupOnClose property This commit makes it possible to set the option to leave the group on close at configuration/deployment time using configuration. See gh-50432 Signed-off-by: Trygve Aasjord --- .../boot/kafka/autoconfigure/KafkaProperties.java | 13 +++++++++++++ .../KafkaStreamsAnnotationDrivenConfiguration.java | 1 + 2 files changed, 14 insertions(+) diff --git a/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaProperties.java b/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaProperties.java index d622810b5fee..9f66725b7d11 100644 --- a/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaProperties.java +++ b/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaProperties.java @@ -812,6 +812,11 @@ public static class Streams { */ private @Nullable String stateDir; + /** + * Whether the consumer should leave the group when stopping Kafka Streams. + */ + private boolean leaveGroupOnClose; + /** * Additional Kafka properties used to configure the streams. */ @@ -885,6 +890,14 @@ public void setStateDir(@Nullable String stateDir) { this.stateDir = stateDir; } + public boolean isLeaveGroupOnClose() { + return this.leaveGroupOnClose; + } + + public void setLeaveGroupOnClose(boolean leaveGroupOnClose) { + this.leaveGroupOnClose = leaveGroupOnClose; + } + public Map getProperties() { return this.properties; } diff --git a/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaStreamsAnnotationDrivenConfiguration.java b/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaStreamsAnnotationDrivenConfiguration.java index 1c7184960063..410d72067254 100644 --- a/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaStreamsAnnotationDrivenConfiguration.java +++ b/module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaStreamsAnnotationDrivenConfiguration.java @@ -101,6 +101,7 @@ public void configure(StreamsBuilderFactoryBean factoryBean) { KafkaProperties.Cleanup cleanup = this.properties.getStreams().getCleanup(); CleanupConfig cleanupConfig = new CleanupConfig(cleanup.isOnStartup(), cleanup.isOnShutdown()); factoryBean.setCleanupConfig(cleanupConfig); + factoryBean.setLeaveGroupOnClose(this.properties.getStreams().isLeaveGroupOnClose()); } @Override From 1383d34d1edb48c5a3adeaed5199c06211f67403 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Nicoll?= Date: Wed, 15 Jul 2026 11:58:00 +0200 Subject: [PATCH 2/2] Polish "Add Kafka Stream's leaveGroupOnClose property" See gh-50432 --- .../autoconfigure/KafkaAutoConfigurationTests.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java b/module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java index e65e217272d1..f4cf275aa63b 100644 --- a/module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java +++ b/module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java @@ -475,6 +475,7 @@ void streamsBuilderFactoryBeanConfigurerIsApplied() { assertThat(factoryBean.isAutoStartup()).isFalse(); assertThat(factoryBean).extracting("cleanupConfig.onStart").isEqualTo(true); assertThat(factoryBean).extracting("cleanupConfig.onStop").isEqualTo(true); + assertThat(factoryBean).extracting("leaveGroupOnClose").isEqualTo(false); factoryBean.addListener(listener); }) .withPropertyValues("spring.kafka.client-id=cid", @@ -696,6 +697,19 @@ void streamsWithCleanupConfig() { }); } + @Test + void streamsWithLeaveGroupOnClose() { + this.contextRunner + .withUserConfiguration(EnableKafkaStreamsConfiguration.class, TestKafkaStreamsConfiguration.class) + .withPropertyValues("spring.application.name=my-test-app", + "spring.kafka.bootstrap-servers=localhost:9092,localhost:9093", + "spring.kafka.streams.auto-startup=false", "spring.kafka.streams.leave-group-on-close=true") + .run((context) -> { + StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean(StreamsBuilderFactoryBean.class); + assertThat(streamsBuilderFactoryBean).extracting("leaveGroupOnClose").isEqualTo(true); + }); + } + @Test void streamsApplicationIdIsMandatory() { this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class).run((context) -> {