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

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

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);
 	}
 }

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<String, ResolvableType> resolvableTypeMap = new TreeMap<>();
 
+	private final StreamFunctionProperties streamFunctionProperties;
+
+	public KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) {
+		this.streamFunctionProperties = streamFunctionProperties;
+	}
+
 	public Map<String, ResolvableType> 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<String> 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<String> 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);
 		}
 	}
 

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",