diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java index 14f69b9f3..ef5d33b1c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java @@ -74,16 +74,13 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto private final String functionName; - private final boolean onlySingleFunction; - private BeanFactory beanFactory; - public KafkaStreamsBindableProxyFactory(ResolvableType type, String functionName, boolean onlySingleFunction) { + public KafkaStreamsBindableProxyFactory(ResolvableType type, String functionName) { super(type.getType().getClass()); this.type = type; this.functionName = functionName; - this.onlySingleFunction = onlySingleFunction; } @Override diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java index c37b1bfc4..e0bf03345 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java @@ -16,13 +16,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; -import org.springframework.beans.factory.config.BeanFactoryPostProcessor; -import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsFunctionProcessor; -import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Conditional; @@ -34,7 +29,6 @@ import org.springframework.context.annotation.Configuration; */ @Configuration @EnableConfigurationProperties(StreamFunctionProperties.class) -@AutoConfigureBefore(BinderFactoryAutoConfiguration.class) public class KafkaStreamsFunctionAutoConfiguration { @Bean @@ -49,27 +43,7 @@ public class KafkaStreamsFunctionAutoConfiguration { @Bean @Conditional(FunctionDetectorCondition.class) - public KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor() { - return new KafkaStreamsFunctionBeanPostProcessor(); - } - - @Bean - @Conditional(FunctionDetectorCondition.class) - public static BeanFactoryPostProcessor implicitFunctionKafkaStreamsBinder(KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor) { - return beanFactory -> { - BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - - for (String s : kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes().keySet()) { - RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( - KafkaStreamsBindableProxyFactory.class); - rootBeanDefinition.getConstructorArgumentValues() - .addGenericArgumentValue(kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes().get(s)); - rootBeanDefinition.getConstructorArgumentValues() - .addGenericArgumentValue(s); - rootBeanDefinition.getConstructorArgumentValues() - .addGenericArgumentValue(kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes().size() == 1); - registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory-" + s, rootBeanDefinition); - } - }; + public KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) { + return new KafkaStreamsFunctionBeanPostProcessor(streamFunctionProperties); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 9047e3fc8..0f9943a86 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; import java.lang.reflect.Method; import java.util.Arrays; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.TreeMap; @@ -25,6 +26,7 @@ import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; +import java.util.stream.Collectors; import java.util.stream.Stream; import org.apache.commons.logging.Log; @@ -39,6 +41,9 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; import org.springframework.util.ClassUtils; @@ -52,9 +57,18 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, private static final Log LOG = LogFactory.getLog(KafkaStreamsFunctionBeanPostProcessor.class); + private static final String[] EXCLUDE_FUNCTIONS = new String[]{"functionRouter", "sendToDlqAndContinue"}; + private ConfigurableListableBeanFactory beanFactory; + private boolean onlySingleFunction; private Map resolvableTypeMap = new TreeMap<>(); + private final StreamFunctionProperties streamFunctionProperties; + + public KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) { + this.streamFunctionProperties = streamFunctionProperties; + } + public Map getResolvableTypes() { return this.resolvableTypeMap; } @@ -66,10 +80,26 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, String[] consumerNames = this.beanFactory.getBeanNamesForType(Consumer.class); String[] biConsumerNames = this.beanFactory.getBeanNamesForType(BiConsumer.class); - Stream.concat( + final Stream concat = Stream.concat( Stream.concat(Stream.of(functionNames), Stream.of(consumerNames)), - Stream.concat(Stream.of(biFunctionNames), Stream.of(biConsumerNames))) + Stream.concat(Stream.of(biFunctionNames), Stream.of(biConsumerNames))); + final List collect = concat.collect(Collectors.toList()); + collect.removeIf(s -> Arrays.stream(EXCLUDE_FUNCTIONS).anyMatch(t -> t.equals(s))); + onlySingleFunction = collect.size() == 1; + collect.stream() .forEach(this::extractResolvableTypes); + + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; + + for (String s : getResolvableTypes().keySet()) { + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + KafkaStreamsBindableProxyFactory.class); + rootBeanDefinition.getConstructorArgumentValues() + .addGenericArgumentValue(getResolvableTypes().get(s)); + rootBeanDefinition.getConstructorArgumentValues() + .addGenericArgumentValue(s); + registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory-" + s, rootBeanDefinition); + } } private void extractResolvableTypes(String key) { @@ -85,12 +115,23 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj); final Class rawClass = resolvableType.getGeneric(0).getRawClass(); if (rawClass == KStream.class || rawClass == KTable.class || rawClass == GlobalKTable.class) { - resolvableTypeMap.put(key, resolvableType); + if (onlySingleFunction) { + resolvableTypeMap.put(key, resolvableType); + } + else { + final String definition = streamFunctionProperties.getDefinition(); + if (definition == null) { + throw new IllegalStateException("Multiple functions found, but function definition property is not set."); + } + else if (definition.contains(key)) { + resolvableTypeMap.put(key, resolvableType); + } + } } } } catch (Exception e) { - LOG.error("Function not found: " + key, e); + LOG.error("Function activation issues while mapping the function: " + key, e); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java index fee4cd9fc..47c378b5a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java @@ -83,6 +83,7 @@ public class MultipleFunctionsInSameAppTests { try (ConfigurableApplicationContext context = app.run( "--server.port=0", "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=process;analyze", "--spring.cloud.stream.bindings.process-in-0.destination=purchases", "--spring.cloud.stream.bindings.process-out-0.destination=coffee", "--spring.cloud.stream.bindings.process-out-1.destination=electronics",