Merge pull request #46675 from nosan

* pr/46675:
  Polish "Use Kafka's StreamsBuilderFactoryBeanConfigurer"
  Use Kafka's StreamsBuilderFactoryBeanConfigurer

Closes gh-46675

Author: snicoll

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

@@ -74,6 +74,8 @@ 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: You can also register an arbitrary number of beans that implement javadoc:org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer[] for more advanced customizations.
+
 
 
 [[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,25 @@ private void applyKafkaConnectionDetailsForStreams(Map<String, Object> propertie
 		KafkaAutoConfiguration.applySslBundle(properties, streams.getSslBundle());
 	}
 
-	// Separate class required to avoid BeanCurrentlyInCreationException
-	static class KafkaStreamsFactoryBeanConfigurer implements InitializingBean {
+	static class KafkaPropertiesStreamsBuilderFactoryBeanConfigurer implements StreamsBuilderFactoryBeanConfigurer {
 
 		private final KafkaProperties properties;
 
-		private final StreamsBuilderFactoryBean factoryBean;
-
-		KafkaStreamsFactoryBeanConfigurer(KafkaProperties properties, StreamsBuilderFactoryBean factoryBean) {
+		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)