diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index c87908840..853397c00 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -53,6 +53,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.ResolvableType; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.config.StreamsBuilderFactoryBean; +import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; import org.springframework.kafka.core.CleanupConfig; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; @@ -150,7 +151,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application @SuppressWarnings({"unchecked"}) protected StreamsBuilderFactoryBean buildStreamsBuilderAndRetrieveConfig(String beanNamePostPrefix, ApplicationContext applicationContext, String inboundName, - KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, + StreamsBuilderFactoryBeanCustomizer customizer) { ConfigurableListableBeanFactory beanFactory = this.applicationContext .getBeanFactory(); @@ -216,6 +218,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application ? new StreamsBuilderFactoryBean(kafkaStreamsConfiguration) : new StreamsBuilderFactoryBean(kafkaStreamsConfiguration, this.cleanupConfig); + if (customizer != null) { + customizer.configure(streamsBuilder); + } streamsBuilder.setAutoStartup(false); BeanDefinition streamsBuilderBeanDefinition = BeanDefinitionBuilder .genericBeanDefinition( diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index ea5b726ee..e5b6f70de 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -60,6 +60,7 @@ import org.springframework.core.env.Environment; import org.springframework.core.env.MapPropertySource; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.kafka.config.KafkaStreamsConfiguration; +import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler; import org.springframework.lang.Nullable; @@ -268,12 +269,13 @@ public class KafkaStreamsBinderSupportAutoConfiguration { KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, KStreamStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter, Collection streamListenerResultAdapters, - ObjectProvider cleanupConfig) { + ObjectProvider cleanupConfig, + ObjectProvider customizerProvider) { return new KafkaStreamsStreamListenerSetupMethodOrchestrator( bindingServiceProperties, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, kafkaStreamsBindingInformationCatalogue, kafkaStreamListenerParameterAdapter, streamListenerResultAdapters, - cleanupConfig.getIfUnique()); + cleanupConfig.getIfUnique(), customizerProvider.getIfUnique()); } @Bean @@ -370,10 +372,12 @@ public class KafkaStreamsBinderSupportAutoConfiguration { KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate, ObjectProvider cleanupConfig, StreamFunctionProperties streamFunctionProperties, - KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, + ObjectProvider customizerProvider) { return new KafkaStreamsFunctionProcessor(bindingServiceProperties, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, kafkaStreamsBindingInformationCatalogue, kafkaStreamsMessageConversionDelegate, - cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties); + cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties, + customizerProvider.getIfUnique()); } @Bean diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 4d1e4ac50..210bc7fc7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -56,6 +56,7 @@ import org.springframework.cloud.stream.function.FunctionConstants; import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; import org.springframework.kafka.config.StreamsBuilderFactoryBean; +import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; import org.springframework.kafka.core.CleanupConfig; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -79,6 +80,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro private BeanFactory beanFactory; private StreamFunctionProperties streamFunctionProperties; private KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties; + StreamsBuilderFactoryBeanCustomizer customizer; public KafkaStreamsFunctionProcessor(BindingServiceProperties bindingServiceProperties, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, @@ -87,7 +89,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate, CleanupConfig cleanupConfig, StreamFunctionProperties streamFunctionProperties, - KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, + StreamsBuilderFactoryBeanCustomizer customizer) { super(bindingServiceProperties, kafkaStreamsBindingInformationCatalogue, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, cleanupConfig); this.bindingServiceProperties = bindingServiceProperties; @@ -97,6 +100,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate; this.streamFunctionProperties = streamFunctionProperties; this.kafkaStreamsBinderConfigurationProperties = kafkaStreamsBinderConfigurationProperties; + this.customizer = customizer; } private Map buildTypeMap(ResolvableType resolvableType, @@ -290,7 +294,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro //Retrieve the StreamsConfig created for this method if available. //Otherwise, create the StreamsBuilderFactory and get the underlying config. if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(functionName)) { - StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(functionName, applicationContext, input, kafkaStreamsBinderConfigurationProperties); + StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(functionName, applicationContext, + input, kafkaStreamsBinderConfigurationProperties, customizer); this.methodStreamsBuilderFactoryBeanMap.put(functionName, streamsBuilderFactoryBean); } try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index 50a5d096c..2a7d19622 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -54,6 +54,7 @@ import org.springframework.core.MethodParameter; import org.springframework.core.ResolvableType; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.kafka.config.StreamsBuilderFactoryBean; +import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; import org.springframework.kafka.core.CleanupConfig; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.util.Assert; @@ -98,6 +99,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr private final Map methodStreamsBuilderFactoryBeanMap = new HashMap<>(); + StreamsBuilderFactoryBeanCustomizer customizer; + KafkaStreamsStreamListenerSetupMethodOrchestrator( BindingServiceProperties bindingServiceProperties, KafkaStreamsExtendedBindingProperties extendedBindingProperties, @@ -105,7 +108,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr KafkaStreamsBindingInformationCatalogue bindingInformationCatalogue, StreamListenerParameterAdapter streamListenerParameterAdapter, Collection listenerResultAdapters, - CleanupConfig cleanupConfig) { + CleanupConfig cleanupConfig, + StreamsBuilderFactoryBeanCustomizer customizer) { super(bindingServiceProperties, bindingInformationCatalogue, extendedBindingProperties, keyValueSerdeResolver, cleanupConfig); this.bindingServiceProperties = bindingServiceProperties; this.kafkaStreamsExtendedBindingProperties = extendedBindingProperties; @@ -113,6 +117,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr this.kafkaStreamsBindingInformationCatalogue = bindingInformationCatalogue; this.streamListenerParameterAdapter = streamListenerParameterAdapter; this.streamListenerResultAdapters = listenerResultAdapters; + this.customizer = customizer; } @Override @@ -244,7 +249,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(method)) { StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(method.getDeclaringClass().getSimpleName() + "-" + method.getName(), applicationContext, - inboundName, null); + inboundName, null, customizer); this.methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderFactoryBean); } try {