From 5454d54faf927bb23dfa3132fa1a3d91acdf7ed0 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 16 Jun 2021 22:56:18 -0400 Subject: [PATCH] Support Kafka Streams binder function composition Composed function defintions can start with a `java.util.function.Function` or `java.util.function.BiFunction` and compose with other functions or consumers. In the case of a consumer, this needs to be the last unit in the function definition. `java.util.function.BiConumer` is not eligibe for function composition. The first component in the definition can be a curried function as well. Adding tests and docs. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1088 Resolves #1091 --- docs/src/main/asciidoc/kafka-streams.adoc | 101 ++++ .../KafkaStreamsFunctionProcessor.java | 124 +++-- .../KafkaStreamsBindableProxyFactory.java | 60 ++- ...KafkaStreamsFunctionAutoConfiguration.java | 6 +- ...KafkaStreamsFunctionBeanPostProcessor.java | 57 ++- .../KafkaStreamsFunctionProcessorInvoker.java | 47 +- .../KafkaStreamsFunctionCompositionTests.java | 468 ++++++++++++++++++ 7 files changed, 790 insertions(+), 73 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 313c5a91b..74e7b0d20 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -303,6 +303,107 @@ In summary, the following table shows the various options that can be used in th * In the case of more than one output in this table, the type simply becomes `KStream[]`. +===== Function composition in Kafka Streams binder + +Kafka Streams binder supports minimal forms of functional composition for linear topologies. +Using the Java functional API support, you can write multiple functions and then compose them on your own using the `andThen` method. +For example, assume that you have the following two functions. + +``` +@Bean +public Function, KStream> foo() { + return input -> input.peek((s, s2) -> {}); +} + +@Bean +public Function, KStream> bar() { + return input -> input.peek((s, s2) -> {}); +} +``` + +Even without the functional composition support in the binder, you can compose these two functions as below. + +``` +@Bean +pubic Funcion, KStream> composed() { + foo().andThen(bar()); +} +``` + +Then you can provide deefinitions of the form `spring.cloud.stream.function.definition=foo;bar;composed`. +With the functional composition support in the binder, you don't need to write this third function in which you are doing explicit function composition. + +You can simply do this instead: + +``` +spring.cloud.stream.function.definition=foo|bar +``` + +You can even do this: + +``` +spring.cloud.stream.function.definition=foo|bar;foo;bar +``` + +The composed function's default binding names in this example becomes `foobar-in-0` and `foobar-out-0`. + +====== Limitations of functional composition in Kafka Streams bincer + +When you have `java.util.function.Function` bean, that can be composed with another function or multiple functions. +The same function bean can be composed with a `java.util.function.Consumer` as well. In this case, consumer is the last component composed. +A function can be composed with multiple functions, then end with a `java.util.function.Consumer` bean as well. + +When composing the beans of type `java.util.function.BiFunction`, the `BiFunction` must be the first function in the definition. +The composed entities must be either of type `java.util.function.Function` or `java.util.funciton.Consumer`. +In other words, you cannot take a `BiFunction` bean and then compose with another `BiFunction`. + +You cannot compose with types of `BiConsumer` or definitions where `Consumer` is the first component. +You cannot also compose with functions where the output is an array (`KStream[]` for branching) unless this is the last component in the definition. + +The very first `Function` of `BiFunction` in the function definition may use a curried form also. +For example, the following is possible. + +``` +@Bean +public Function, Function, KStream>> curriedFoo() { + return a -> b -> + a.join(b, (value1, value2) -> value1 + value2); +} + +@Bean +public Function, KStream> bar() { + return input -> input.mapValues(value -> value + "From-anotherFooFunc"); +} +``` + +and the function definition could be `curriedFoo|bar`. +Behind the scenes, the binder will create two input bindings for the curried function, and an output binding based on the final function in the definition. +The default input bindings in this case are going to be `curriedFoobar-in-0` and `curriedFoobar-in-1`. +The default output binding for this example becomes `curriedFoobar-out-0`. + +====== Special note on using `KTable` as output in function composition + +When using function composition, for intermediate functions, you can use `KTable` as output. +For instance, lets say you have the following two functions. + +``` +@Bean +public Function, KTable> foo() { + return KStream::toTable; + }; +} + +@Bean +public Function, KStream> bar() { + return KTable::toStream; +} +``` + +You can compose them as `foo|bar` although foo's output is `KTable`. +In normal case, when you use `foo` as standalone, this will not work, as the binder does not support `KTable` as the final output. +Note that in the example above, bar's output is still a `KStream`. +We are only able to use `foo` which has a `KTable` output, since we are composing with another function that has `KStream` as its output. + ==== Imperative programming model. Starting with `3.1.0` version of the binder, we recommend using the functional programming model described above for Kafka Streams binder based applications. 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 4adc4c34e..48e502e29 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 @@ -216,6 +216,23 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro } } + private ResolvableType checkOutboundForComposedFunctions( + ResolvableType outputResolvableType) { + + ResolvableType currentOutputGeneric; + + if (outputResolvableType.getRawClass() != null && outputResolvableType.getRawClass().isAssignableFrom(BiFunction.class)) { + currentOutputGeneric = outputResolvableType.getGeneric(2); + } + else { + currentOutputGeneric = outputResolvableType.getGeneric(1); + } + while (currentOutputGeneric.getRawClass() != null && functionOrConsumerFound(currentOutputGeneric)) { + currentOutputGeneric = currentOutputGeneric.getGeneric(1); + } + return currentOutputGeneric; + } + /** * This method must be kept stateless. In the case of multiple function beans in an application, * isolated {@link KafkaStreamsBindableProxyFactory} instances are passed in separately for those functions. If the @@ -228,10 +245,21 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro */ @SuppressWarnings({ "unchecked", "rawtypes" }) public void setupFunctionInvokerForKafkaStreams(ResolvableType resolvableType, String functionName, - KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory, Method method) { + KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory, Method method, + ResolvableType outputResolvableType, + String... composedFunctionNames) { final Map resolvableTypes = buildTypeMap(resolvableType, kafkaStreamsBindableProxyFactory, method, functionName); - ResolvableType outboundResolvableType = resolvableTypes.remove(OUTBOUND); + + ResolvableType outboundResolvableType; + if (outputResolvableType != null) { + outboundResolvableType = checkOutboundForComposedFunctions(outputResolvableType); + resolvableTypes.remove(OUTBOUND); + } + else { + outboundResolvableType = resolvableTypes.remove(OUTBOUND); + } + Object[] adaptedInboundArguments = adaptAndRetrieveInboundArguments(resolvableTypes, functionName); try { if (resolvableType.getRawClass() != null && resolvableType.getRawClass().equals(Consumer.class)) { @@ -258,17 +286,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro else { result = ((Function) bean).apply(adaptedInboundArguments[0]); } - int i = 1; - while (result instanceof Function || result instanceof Consumer) { - if (result instanceof Function) { - result = ((Function) result).apply(adaptedInboundArguments[i]); - } - else { - ((Consumer) result).accept(adaptedInboundArguments[i]); - result = null; - } - i++; - } + result = handleCurriedFunctions(adaptedInboundArguments, result); if (result != null) { final Set outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs()); final Iterator outboundDefinitionIterator = outputs.iterator(); @@ -286,25 +304,26 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro } } else { - Object result; + Object result = null; if (resolvableType.getRawClass() != null && resolvableType.getRawClass().equals(BiFunction.class)) { - BiFunction biFunction = (BiFunction) beanFactory.getBean(functionName); - result = biFunction.apply(adaptedInboundArguments[0], adaptedInboundArguments[1]); - } - else { - Function function = (Function) beanFactory.getBean(functionName); - result = function.apply(adaptedInboundArguments[0]); - } - int i = 1; - while (result instanceof Function || result instanceof Consumer) { - if (result instanceof Function) { - result = ((Function) result).apply(adaptedInboundArguments[i]); + if (composedFunctionNames != null && composedFunctionNames.length > 0) { + result = handleComposedFunctions(adaptedInboundArguments, result, composedFunctionNames); } else { - ((Consumer) result).accept(adaptedInboundArguments[i]); - result = null; + BiFunction biFunction = (BiFunction) beanFactory.getBean(functionName); + result = biFunction.apply(adaptedInboundArguments[0], adaptedInboundArguments[1]); + result = handleCurriedFunctions(adaptedInboundArguments, result); + } + } + else { + if (composedFunctionNames != null && composedFunctionNames.length > 0) { + result = handleComposedFunctions(adaptedInboundArguments, result, composedFunctionNames); + } + else { + Function function = (Function) beanFactory.getBean(functionName); + result = function.apply(adaptedInboundArguments[0]); + result = handleCurriedFunctions(adaptedInboundArguments, result); } - i++; } if (result != null) { final Set outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs()); @@ -329,6 +348,55 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro } } + @SuppressWarnings({"unchecked"}) + private Object handleComposedFunctions(Object[] adaptedInboundArguments, Object result, String... composedFunctionNames) { + Object bean = beanFactory.getBean(composedFunctionNames[0]); + if (BiFunction.class.isAssignableFrom(bean.getClass())) { + result = ((BiFunction) bean).apply(adaptedInboundArguments[0], adaptedInboundArguments[1]); + } + else if (Function.class.isAssignableFrom(bean.getClass())) { + result = ((Function) bean).apply(adaptedInboundArguments[0]); + } + // If the return is a curried function, apply it + result = handleCurriedFunctions(adaptedInboundArguments, result); + // Apply composed functions + return applyComposedFunctions(result, composedFunctionNames); + } + + @SuppressWarnings({"unchecked"}) + private Object applyComposedFunctions(Object result, String[] composedFunctionNames) { + for (int i = 1; i < composedFunctionNames.length; i++) { + final Object bean = beanFactory.getBean(composedFunctionNames[i]); + if (Consumer.class.isAssignableFrom(bean.getClass())) { + ((Consumer) bean).accept(result); + result = null; + } + else if (Function.class.isAssignableFrom(bean.getClass())) { + result = ((Function) bean).apply(result); + } + else { + throw new IllegalStateException("You can only compose functions of type either java.util.function.Function or java.util.function.Consumer."); + } + } + return result; + } + + @SuppressWarnings({"unchecked"}) + private Object handleCurriedFunctions(Object[] adaptedInboundArguments, Object result) { + int i = 1; + while (result instanceof Function || result instanceof Consumer) { + if (result instanceof Function) { + result = ((Function) result).apply(adaptedInboundArguments[i]); + } + else { + ((Consumer) result).accept(adaptedInboundArguments[i]); + result = null; + } + i++; + } + return result; + } + private void handleSingleKStreamOutbound(Map resolvableTypes, ResolvableType outboundResolvableType, KStream result, Iterator outboundDefinitionIterator) { if (outboundDefinitionIterator.hasNext()) { 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 2a49cfd14..4a87bf488 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 @@ -49,16 +49,16 @@ import org.springframework.util.CollectionUtils; /** * Kafka Streams specific target bindings proxy factory. See {@link AbstractBindableProxyFactory} for more details. - * + *

* Targets bound by this factory: - * + *

* {@link KStream} * {@link KTable} * {@link GlobalKTable} - * + *

* This class looks at the Function bean's return signature as {@link ResolvableType} and introspect the individual types, * binding them on the way. - * + *

* All types on the {@link ResolvableType} are bound except for KStream[] array types on the outbound, which will be * deferred for binding at a later stage. The reason for doing that is because in this class, we don't have any way to know * the actual size in the returned array. That has to wait until the function is invoked and we get a result. @@ -71,17 +71,17 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto @Autowired private StreamFunctionProperties streamFunctionProperties; - private final ResolvableType type; + private ResolvableType[] types; - private final Method method; + private Method method; private final String functionName; private BeanFactory beanFactory; - public KafkaStreamsBindableProxyFactory(ResolvableType type, String functionName, Method method) { - super(type.getType().getClass()); - this.type = type; + public KafkaStreamsBindableProxyFactory(ResolvableType[] types, String functionName, Method method) { + super(types[0].getType().getClass()); + this.types = types; this.functionName = functionName; this.method = method; } @@ -93,10 +93,10 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto "'bindingTargetFactories' cannot be empty"); int resolvableTypeDepthCounter = 0; - boolean isKafkaStreamsType = this.type.getRawClass().isAssignableFrom(KStream.class) || - this.type.getRawClass().isAssignableFrom(KTable.class) || - this.type.getRawClass().isAssignableFrom(GlobalKTable.class); - ResolvableType argument = isKafkaStreamsType ? this.type : this.type.getGeneric(resolvableTypeDepthCounter++); + boolean isKafkaStreamsType = this.types[0].getRawClass().isAssignableFrom(KStream.class) || + this.types[0].getRawClass().isAssignableFrom(KTable.class) || + this.types[0].getRawClass().isAssignableFrom(GlobalKTable.class); + ResolvableType argument = isKafkaStreamsType ? this.types[0] : this.types[0].getGeneric(resolvableTypeDepthCounter++); List inputBindings = buildInputBindings(); Iterator iterator = inputBindings.iterator(); String next = iterator.next(); @@ -112,18 +112,21 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto } } // Normal functional bean - if (this.type.getRawClass() != null && - (this.type.getRawClass().isAssignableFrom(BiFunction.class) || - this.type.getRawClass().isAssignableFrom(BiConsumer.class))) { - argument = this.type.getGeneric(resolvableTypeDepthCounter++); + if (this.types[0].getRawClass() != null && + (this.types[0].getRawClass().isAssignableFrom(BiFunction.class) || + this.types[0].getRawClass().isAssignableFrom(BiConsumer.class))) { + argument = this.types[0].getGeneric(resolvableTypeDepthCounter++); next = iterator.next(); bindInput(argument, next); } - ResolvableType outboundArgument = this.type.getGeneric(resolvableTypeDepthCounter); - + ResolvableType outboundArgument; if (method != null) { outboundArgument = ResolvableType.forMethodReturnType(method); } + else { + outboundArgument = this.types[0].getGeneric(resolvableTypeDepthCounter); + } + while (isAnotherFunctionOrConsumerFound(outboundArgument)) { //The function is a curried function. We should introspect the partial function chain hierarchy. argument = outboundArgument.getGeneric(0); @@ -132,7 +135,16 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto outboundArgument = outboundArgument.getGeneric(1); } - if (outboundArgument.getRawClass() != null && (!outboundArgument.isArray() && + + final int lastTypeIndex = this.types.length - 1; + if (this.types.length > 1 && this.types[lastTypeIndex] != null && this.types[lastTypeIndex].getRawClass() != null) { + if (this.types[lastTypeIndex].getRawClass().isAssignableFrom(Function.class) || + this.types[lastTypeIndex].getRawClass().isAssignableFrom(Consumer.class)) { + outboundArgument = this.types[lastTypeIndex].getGeneric(1); + } + } + + if (outboundArgument != null && outboundArgument.getRawClass() != null && (!outboundArgument.isArray() && outboundArgument.getRawClass().isAssignableFrom(KStream.class))) { // if the type is array, we need to do a late binding as we don't know the number of // output bindings at this point in the flow. @@ -181,9 +193,9 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto inputs.addAll(inputBindings); return inputs; } - int numberOfInputs = this.type.getRawClass() != null && - (this.type.getRawClass().isAssignableFrom(BiFunction.class) || - this.type.getRawClass().isAssignableFrom(BiConsumer.class)) ? 2 : getNumberOfInputs(); + int numberOfInputs = this.types[0].getRawClass() != null && + (this.types[0].getRawClass().isAssignableFrom(BiFunction.class) || + this.types[0].getRawClass().isAssignableFrom(BiConsumer.class)) ? 2 : getNumberOfInputs(); // For @Component style beans. if (method != null) { @@ -213,7 +225,7 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto private int getNumberOfInputs() { int numberOfInputs = 1; - ResolvableType arg1 = this.type.getGeneric(1); + ResolvableType arg1 = this.types[0].getGeneric(1); while (isAnotherFunctionOrConsumerFound(arg1)) { arg1 = arg1.getGeneric(1); 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 47e957e1f..10588ccc7 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 @@ -36,9 +36,11 @@ public class KafkaStreamsFunctionAutoConfiguration { public KafkaStreamsFunctionProcessorInvoker kafkaStreamsFunctionProcessorInvoker( KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor, KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, - KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories) { + KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories, + StreamFunctionProperties streamFunctionProperties) { return new KafkaStreamsFunctionProcessorInvoker(kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes(), - kafkaStreamsFunctionProcessor, kafkaStreamsBindableProxyFactories, kafkaStreamsFunctionBeanPostProcessor.getMethods()); + kafkaStreamsFunctionProcessor, kafkaStreamsBindableProxyFactories, kafkaStreamsFunctionBeanPostProcessor.getMethods(), + streamFunctionProperties); } @Bean 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 542a6a0ad..71cf073fd 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 @@ -48,6 +48,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; import org.springframework.util.ClassUtils; +import org.springframework.util.StringUtils; /** * @author Soby Chacko @@ -103,17 +104,53 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, 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); - rootBeanDefinition.getConstructorArgumentValues() - .addGenericArgumentValue(getMethods().get(s)); - registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory-" + s, rootBeanDefinition); + final String definition = streamFunctionProperties.getDefinition(); + final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{}; + + if (functionUnits.length == 0) { + for (String s : getResolvableTypes().keySet()) { + ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(s)}; + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + KafkaStreamsBindableProxyFactory.class); + registerKakaStreamsProxyFactory(registry, s, resolvableTypes, rootBeanDefinition); + } } + else { + for (String functionUnit : functionUnits) { + if (functionUnit.contains("|")) { + final String[] composedFunctions = functionUnit.split("\\|"); + String derivedNameFromComposed = ""; + ResolvableType[] resolvableTypes = new ResolvableType[composedFunctions.length]; + + int i = 0; + for (String split : composedFunctions) { + derivedNameFromComposed = derivedNameFromComposed.concat(split); + resolvableTypes[i++] = getResolvableTypes().get(split); + } + + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + KafkaStreamsBindableProxyFactory.class); + registerKakaStreamsProxyFactory(registry, derivedNameFromComposed, resolvableTypes, rootBeanDefinition); + } + else { + + ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(functionUnit)}; + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + KafkaStreamsBindableProxyFactory.class); + registerKakaStreamsProxyFactory(registry, functionUnit, resolvableTypes, rootBeanDefinition); + } + } + } + } + + private void registerKakaStreamsProxyFactory(BeanDefinitionRegistry registry, String s, ResolvableType[] resolvableTypes, RootBeanDefinition rootBeanDefinition) { + rootBeanDefinition.getConstructorArgumentValues() + .addGenericArgumentValue(resolvableTypes); + rootBeanDefinition.getConstructorArgumentValues() + .addGenericArgumentValue(s); + rootBeanDefinition.getConstructorArgumentValues() + .addGenericArgumentValue(getMethods().get(s)); + registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory-" + s, rootBeanDefinition); } private void extractResolvableTypes(String key) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java index 4b6d1be6c..acc2d9173 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java @@ -24,10 +24,11 @@ import java.util.Optional; import javax.annotation.PostConstruct; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsFunctionProcessor; +import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; +import org.springframework.util.StringUtils; /** - * * @author Soby Chacko * @since 2.1.0 */ @@ -37,23 +38,51 @@ public class KafkaStreamsFunctionProcessorInvoker { private final Map resolvableTypeMap; private final KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories; private final Map methods; + private final StreamFunctionProperties streamFunctionProperties; public KafkaStreamsFunctionProcessorInvoker(Map resolvableTypeMap, - KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, - KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories, - Map methods) { + KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, + KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories, + Map methods, StreamFunctionProperties streamFunctionProperties) { this.kafkaStreamsFunctionProcessor = kafkaStreamsFunctionProcessor; this.resolvableTypeMap = resolvableTypeMap; this.kafkaStreamsBindableProxyFactories = kafkaStreamsBindableProxyFactories; this.methods = methods; + this.streamFunctionProperties = streamFunctionProperties; } @PostConstruct void invoke() { - resolvableTypeMap.forEach((key, value) -> { - Optional proxyFactory = - Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(key)).findFirst(); - this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key, proxyFactory.get(), methods.get(key)); - }); + final String definition = streamFunctionProperties.getDefinition(); + final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{}; + + if (functionUnits.length == 0) { + resolvableTypeMap.forEach((key, value) -> { + Optional proxyFactory = + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(key)).findFirst(); + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key, proxyFactory.get(), methods.get(key), null); + }); + } + + for (String functionUnit : functionUnits) { + if (functionUnit.contains("|")) { + final String[] composedFunctions = functionUnit.split("\\|"); + String[] derivedNameFromComposed = new String[]{""}; + for (String split : composedFunctions) { + derivedNameFromComposed[0] = derivedNameFromComposed[0].concat(split); + } + Optional proxyFactory = + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(derivedNameFromComposed[0])).findFirst(); + + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]), + derivedNameFromComposed[0], proxyFactory.get(), methods.get(derivedNameFromComposed[0]), resolvableTypeMap.get(composedFunctions[composedFunctions.length - 1]), composedFunctions); + } + else { + Optional proxyFactory = + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(functionUnit)).findFirst(); + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(functionUnit), functionUnit, + proxyFactory.get(), methods.get(functionUnit), null); + } + } } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java new file mode 100644 index 000000000..f3716d414 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java @@ -0,0 +1,468 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BiFunction; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.kstream.ForeachAction; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.util.Assert; + +import static org.assertj.core.api.Assertions.assertThat; + +public class KafkaStreamsFunctionCompositionTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "fooFuncanotherFooFunc-out-0", "bar"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + private static Consumer consumer; + + private static final CountDownLatch countDownLatch1 = new CountDownLatch(1); + private static final CountDownLatch countDownLatch2 = new CountDownLatch(1); + private static final CountDownLatch countDownLatch3 = new CountDownLatch(2); + + @BeforeClass + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("fn-composition-group", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "fooFuncanotherFooFunc-out-0", "bar"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testBasicFunctionCompositionWithDefaultDestination() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig1.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooFunc|anotherFooFunc;anotherProcess", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("fooFuncanotherFooFunc-in-0"); + template.sendDefault("foobar!!"); + //Verify non-composed funcions can be run standalone with composed function chains, i.e foo|bar;buzz + template.setDefaultTopic("anotherProcess-in-0"); + template.sendDefault("this is crazy!!!"); + Thread.sleep(1000); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "fooFuncanotherFooFunc-out-0"); + assertThat(cr.value().contains("foobar!!")).isTrue(); + + Assert.isTrue(countDownLatch1.await(5, TimeUnit.SECONDS), "anotherProcess consumer didn't trigger."); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testBasicFunctionCompositionWithDestinaion() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig1.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooFunc|anotherFooFunc;anotherProcess", + "--spring.cloud.stream.bindings.fooFuncanotherFooFunc-in-0.destination=foo", + "--spring.cloud.stream.bindings.fooFuncanotherFooFunc-out-0.destination=bar", + "--spring.cloud.stream.bindings.anotherProcess-in-0.destination=buzz", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo"); + template.sendDefault("foobar!!"); + template.setDefaultTopic("buzz"); + template.sendDefault("this is crazy!!!"); + Thread.sleep(1000); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "bar"); + assertThat(cr.value().contains("foobar!!")).isTrue(); + + Assert.isTrue(countDownLatch1.await(5, TimeUnit.SECONDS), "anotherProcess consumer didn't trigger."); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testFunctionToConsumerComposition() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig2.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooFunc|anotherProcess", + "--spring.cloud.stream.bindings.fooFuncanotherProcess-in-0.destination=foo", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo"); + template.sendDefault("foobar!!"); + + Thread.sleep(1000); + + Assert.isTrue(countDownLatch2.await(5, TimeUnit.SECONDS), "anotherProcess consumer didn't trigger."); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testBiFunctionToConsumerComposition() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig3.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooBiFunc|anotherProcess", + "--spring.cloud.stream.bindings.fooBiFuncanotherProcess-in-0.destination=foo", + "--spring.cloud.stream.bindings.fooBiFuncanotherProcess-in-1.destination=foo-1", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo"); + template.sendDefault("foobar!!"); + + template.setDefaultTopic("foo-1"); + template.sendDefault("another foobar!!"); + + Thread.sleep(1000); + + Assert.isTrue(countDownLatch3.await(5, TimeUnit.SECONDS), "anotherProcess consumer didn't trigger."); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testChainedFunctionsAsComposed() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig4.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooBiFunc|anotherFooFunc|yetAnotherFooFunc|lastFunctionInChain", + "--spring.cloud.stream.function.bindings.fooBiFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-in-0=input1", + "--spring.cloud.stream.function.bindings.fooBiFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-in-1=input2", + "--spring.cloud.stream.function.bindings.fooBiFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-out-0=output", + "--spring.cloud.stream.bindings.input1.destination=my-foo-1", + "--spring.cloud.stream.bindings.input2.destination=my-foo-2", + "--spring.cloud.stream.bindings.output.destination=bar", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + + template.setDefaultTopic("my-foo-2"); + template.sendDefault("foo-1", "foo2"); + + template.setDefaultTopic("my-foo-1"); + template.sendDefault("foo-1", "foo1"); + + Thread.sleep(1000); + + final ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + assertThat(records.iterator().hasNext()).isTrue(); + assertThat(records.iterator().next().value().equals("foo1foo2From-anotherFooFuncFrom-yetAnotherFooFuncFrom-lastFunctionInChain")).isTrue(); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testFirstFunctionCurriedThenComposeWithOtherFunctions() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig5.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=curriedFunc|anotherFooFunc|yetAnotherFooFunc|lastFunctionInChain", + "--spring.cloud.stream.function.bindings.curriedFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-in-0=input1", + "--spring.cloud.stream.function.bindings.curriedFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-in-1=input2", + "--spring.cloud.stream.function.bindings.curriedFuncanotherFooFuncyetAnotherFooFunclastFunctionInChain-out-0=output", + "--spring.cloud.stream.bindings.input1.destination=my-foo-1", + "--spring.cloud.stream.bindings.input2.destination=my-foo-2", + "--spring.cloud.stream.bindings.output.destination=bar", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + + template.setDefaultTopic("my-foo-2"); + template.sendDefault("foo-1", "foo2"); + + Thread.sleep(1000); + + template.setDefaultTopic("my-foo-1"); + template.sendDefault("foo-1", "foo1"); + + Thread.sleep(1000); + + final ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + assertThat(records.iterator().hasNext()).isTrue(); + assertThat(records.iterator().next().value().equals("foo1foo2From-anotherFooFuncFrom-yetAnotherFooFuncFrom-lastFunctionInChain")).isTrue(); + } + finally { + pf.destroy(); + } + } + } + + @Test + public void testFunctionToConsumerCompositionWithFunctionProducesKTable() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionCompositionConfig6.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=fooFunc|anotherProcess", + "--spring.cloud.stream.bindings.fooFuncanotherProcess-in-0.destination=foo", + "--spring.cloud.stream.bindings.fooFuncanotherProcess-out-0.destination=bar", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo"); + template.sendDefault("foo", "foobar!!"); + + Thread.sleep(1000); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "bar"); + assertThat(cr.value().contains("foobar!!")).isTrue(); + } + finally { + pf.destroy(); + } + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig1 { + + @Bean + public Function, KStream> fooFunc() { + return input -> input.peek((s, s2) -> { + System.out.println("hello: " + s2); + }); + } + + @Bean + public Function, KStream> anotherFooFunc() { + return input -> input.peek((s, s2) -> System.out.println("hello Foo: " + s2)); + } + + @Bean + public java.util.function.Consumer> anotherProcess() { + return c -> c.foreach((s, s2) -> { + System.out.println("s2s2s2::" + s2); + countDownLatch1.countDown(); + }); + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig2 { + + @Bean + public Function, KStream> fooFunc() { + return input -> input.peek((s, s2) -> { + System.out.println("hello: " + s2); + }); + } + + @Bean + public java.util.function.Consumer> anotherProcess() { + return c -> c.foreach((s, s2) -> { + System.out.println("s2s2s2::" + s2); + countDownLatch2.countDown(); + }); + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig3 { + + @Bean + public BiFunction, KStream, KStream> fooBiFunc() { + return KStream::merge; + } + + @Bean + public java.util.function.Consumer> anotherProcess() { + return c -> c.foreach((s, s2) -> { + System.out.println("s2s2s2::" + s2); + countDownLatch3.countDown(); + }); + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig4 { + + @Bean + public BiFunction, KTable, KStream> fooBiFunc() { + return (a, b) -> a.join(b, (value1, value2) -> value1 + value2); + } + + @Bean + public Function, KStream> anotherFooFunc() { + return input -> input.mapValues(value -> value + "From-anotherFooFunc"); + } + + @Bean + public Function, KStream> yetAnotherFooFunc() { + return input -> input.mapValues(value -> value + "From-yetAnotherFooFunc"); + } + + @Bean + public Function, KStream> lastFunctionInChain() { + return input -> input.mapValues(value -> value + "From-lastFunctionInChain"); + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig5 { + + @Bean + public Function, Function, KStream>> curriedFunc() { + return a -> b -> + a.join(b, (value1, value2) -> value1 + value2); + } + + @Bean + public Function, KStream> anotherFooFunc() { + return input -> input.mapValues(value -> value + "From-anotherFooFunc"); + } + + @Bean + public Function, KStream> yetAnotherFooFunc() { + return input -> input.mapValues(value -> value + "From-yetAnotherFooFunc"); + } + + @Bean + public Function, KStream> lastFunctionInChain() { + return input -> input.mapValues(value -> value + "From-lastFunctionInChain"); + } + } + + @EnableAutoConfiguration + public static class FunctionCompositionConfig6 { + + @Bean + public Function, KTable> fooFunc() { + return ks -> { + ks.foreach(new ForeachAction() { + @Override + public void apply(String key, String value) { + System.out.println(); + } + }); + return ks.toTable(); + }; + } + + @Bean + public Function, KStream> anotherProcess() { + return KTable::toStream; + } + } +}