From 28a02cda4f3ca26652560cc7b93e691aadb741b8 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 24 Oct 2019 01:00:48 -0400 Subject: [PATCH] Multiple functions and definition property In order to make Kafka Streams binder based function apps more consistent with the wider functional support in ScSt, it should require the proprety spring.cloud.stream.fucntion.definition to signal which functions to activate. Resolves #783 --- .../KafkaStreamsBindableProxyFactory.java | 5 +- ...KafkaStreamsFunctionAutoConfiguration.java | 30 +----------- ...KafkaStreamsFunctionBeanPostProcessor.java | 49 +++++++++++++++++-- .../MultipleFunctionsInSameAppTests.java | 1 + 4 files changed, 49 insertions(+), 36 deletions(-) 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",