Use Kafka's StreamsBuilderFactoryBeanConfigurer

This commit removes Boot's StreamBuilderFactoryBeanCustomizer and
updates KafkaStreamsAnnotationDrivenConfiguration to use
Kafka's StreamsBuilderFactoryBeanConfigurer instead of the
nested class KafkaStreamsFactoryBeanConfigurer.

This aligns more closely with how configuration is typically applied.

See gh-46675

Signed-off-by: Dmytro Nosan <[email protected]>

Author: nosan

documentation/spring-boot-docs/src/docs/antora/modules/reference/pages/messaging/kafka.adoc

@@ -74,6 +74,7 @@ include-code::MyKafkaStreamsConfiguration[]
 By default, the streams managed by the javadoc:org.apache.kafka.streams.StreamsBuilder[] object are started automatically.
 You can customize this behavior using the configprop:spring.kafka.streams.auto-startup[] property.
 
+TIP: For advanced configuration, the arbitrary javadoc:org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer[] beans can be registered to configure the javadoc:org.springframework.kafka.config.StreamsBuilderFactoryBean[] bean before the javadoc:org.apache.kafka.streams.StreamsBuilder[] bean is initialized.
 
 
 [[messaging.kafka.additional-properties]]

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaStreamsAnnotationDrivenConfiguration.java

@@ -22,19 +22,18 @@
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.source.InvalidConfigurationPropertyValueException;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.Ordered;
 import org.springframework.core.env.Environment;
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
 import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.config.StreamsBuilderFactoryBean;
+import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
 import org.springframework.kafka.core.CleanupConfig;
 
 /**
@@ -76,11 +75,8 @@ KafkaStreamsConfiguration defaultKafkaStreamsConfig(Environment environment,
 	}
 
 	@Bean
-	KafkaStreamsFactoryBeanConfigurer kafkaStreamsFactoryBeanConfigurer(
-			@Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME) StreamsBuilderFactoryBean factoryBean,
-			ObjectProvider<StreamsBuilderFactoryBeanCustomizer> customizers) {
-		customizers.orderedStream().forEach((customizer) -> customizer.customize(factoryBean));
-		return new KafkaStreamsFactoryBeanConfigurer(this.properties, factoryBean);
+	StreamsBuilderFactoryBeanConfigurer kafkaPropertiesStreamsBuilderFactoryBeanConfigurer() {
+		return new KafkaPropertiesStreamsBuilderFactoryBeanConfigurer(this.properties);
 	}
 
 	private void applyKafkaConnectionDetailsForStreams(Map<String, Object> properties,
@@ -91,24 +87,26 @@ private void applyKafkaConnectionDetailsForStreams(Map<String, Object> propertie
 		KafkaAutoConfiguration.applySslBundle(properties, streams.getSslBundle());
 	}
 
-	// Separate class required to avoid BeanCurrentlyInCreationException
-	static class KafkaStreamsFactoryBeanConfigurer implements InitializingBean {
+	private static final class KafkaPropertiesStreamsBuilderFactoryBeanConfigurer
+			implements StreamsBuilderFactoryBeanConfigurer {
 
 		private final KafkaProperties properties;
 
-		private final StreamsBuilderFactoryBean factoryBean;
-
-		KafkaStreamsFactoryBeanConfigurer(KafkaProperties properties, StreamsBuilderFactoryBean factoryBean) {
+		private KafkaPropertiesStreamsBuilderFactoryBeanConfigurer(KafkaProperties properties) {
 			this.properties = properties;
-			this.factoryBean = factoryBean;
 		}
 
 		@Override
-		public void afterPropertiesSet() {
-			this.factoryBean.setAutoStartup(this.properties.getStreams().isAutoStartup());
+		public void configure(StreamsBuilderFactoryBean factoryBean) {
+			factoryBean.setAutoStartup(this.properties.getStreams().isAutoStartup());
 			KafkaProperties.Cleanup cleanup = this.properties.getStreams().getCleanup();
 			CleanupConfig cleanupConfig = new CleanupConfig(cleanup.isOnStartup(), cleanup.isOnShutdown());
-			this.factoryBean.setCleanupConfig(cleanupConfig);
+			factoryBean.setCleanupConfig(cleanupConfig);
+		}
+
+		@Override
+		public int getOrder() {
+			return Ordered.HIGHEST_PRECEDENCE;
 		}
 
 	}

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/StreamsBuilderFactoryBeanCustomizer.java

@@ -26,10 +26,10 @@
 import org.springframework.boot.kafka.autoconfigure.DefaultKafkaConsumerFactoryCustomizer;
 import org.springframework.boot.kafka.autoconfigure.DefaultKafkaProducerFactoryCustomizer;
 import org.springframework.boot.kafka.autoconfigure.KafkaAutoConfiguration;
-import org.springframework.boot.kafka.autoconfigure.StreamsBuilderFactoryBeanCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.config.StreamsBuilderFactoryBean;
+import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.MicrometerConsumerListener;
@@ -75,7 +75,7 @@ private <K, V> void addListener(DefaultKafkaProducerFactory<K, V> factory, Meter
 	static class KafkaStreamsMetricsConfiguration {
 
 		@Bean
-		StreamsBuilderFactoryBeanCustomizer kafkaStreamsMetrics(MeterRegistry meterRegistry) {
+		StreamsBuilderFactoryBeanConfigurer kafkaStreamsMetrics(MeterRegistry meterRegistry) {
 			return (factoryBean) -> factoryBean.addListener(new KafkaStreamsMicrometerListener(meterRegistry));
 		}
 

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/metrics/KafkaMetricsAutoConfiguration.java

@@ -69,6 +69,8 @@
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.config.StreamsBuilderFactoryBean;
+import org.springframework.kafka.config.StreamsBuilderFactoryBean.Listener;
+import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
 import org.springframework.kafka.core.CleanupConfig;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -458,6 +460,29 @@ void streamsProperties() {
 			});
 	}
 
+	@Test
+	void streamsBuilderFactoryBeanConfigurerIsApplied() {
+		Listener listener = mock(Listener.class);
+		this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class)
+			// user's StreamsBuilderFactoryBeanConfigurer must be invoked after the
+			// default one
+			.withBean(StreamsBuilderFactoryBeanConfigurer.class, () -> (factoryBean) -> {
+				assertThat(factoryBean.isAutoStartup()).isFalse();
+				assertThat(factoryBean).extracting("cleanupConfig.onStart").isEqualTo(true);
+				assertThat(factoryBean).extracting("cleanupConfig.onStop").isEqualTo(true);
+				factoryBean.addListener(listener);
+			})
+			.withPropertyValues("spring.kafka.client-id=cid",
+					"spring.kafka.bootstrap-servers=localhost:9092,localhost:9093", "spring.application.name=appName",
+					"spring.kafka.streams.auto-startup=false", "spring.kafka.streams.cleanup.on-shutdown=true",
+					"spring.kafka.streams.cleanup.on-startup=true")
+			.run((context) -> {
+				assertThat(context).hasSingleBean(StreamsBuilderFactoryBean.class);
+				StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean(StreamsBuilderFactoryBean.class);
+				assertThat(streamsBuilderFactoryBean.getListeners()).hasSize(1);
+			});
+	}
+
 	@Test
 	void connectionDetailsAreAppliedToStreams() {
 		this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class)