SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support Kafka Streams binder function composition

Browse Source 
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
This commit is contained in:
Soby Chacko
2021-06-16 22:56:18 -04:00
committed by Oleg Zhurakousky
parent 4d68282937
commit 5454d54faf
7 changed files with 790 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
docs/src/main/asciidoc/kafka-streams.adoc
View File

@@ -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<String, String>, KStream<String, String>> foo() {
return input -> input.peek((s, s2) -> {});
}
@Bean
public Function<KStream<String, String>, KStream<String, Long>> 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<String, String>, KStream<String, Long>> 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<KStream<String, String>, Function<KTable<String, String>, KStream<String, String>>> curriedFoo() {
return a -> b ->
a.join(b, (value1, value2) -> value1 + value2);
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> 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<KStream<String, String>, KTable<String, String>> foo() {
return KStream::toTable;
};
}
@Bean
public Function<KTable<String, String>, KStream<String, String>> 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.

124
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java
View File

@@ -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<String, ResolvableType> 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<String> outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs());
final Iterator<String> 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<Object, Object, Object> biFunction = (BiFunction) beanFactory.getBean(functionName);
result = biFunction.apply(adaptedInboundArguments[0], adaptedInboundArguments[1]);
}
else {
Function<Object, Object> 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<Object, Object, Object> 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<Object, Object> function = (Function) beanFactory.getBean(functionName);
result = function.apply(adaptedInboundArguments[0]);
result = handleCurriedFunctions(adaptedInboundArguments, result);
}
i++;
}
if (result != null) {
final Set<String> 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<Object, Object, Object>) bean).apply(adaptedInboundArguments[0], adaptedInboundArguments[1]);
}
else if (Function.class.isAssignableFrom(bean.getClass())) {
result = ((Function<Object, Object>) 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<Object>) bean).accept(result);
result = null;
}
else if (Function.class.isAssignableFrom(bean.getClass())) {
result = ((Function<Object, Object>) 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<Object, Object>) result).apply(adaptedInboundArguments[i]);
}
else {
((Consumer<Object>) result).accept(adaptedInboundArguments[i]);
result = null;
}
i++;
}
return result;
}
private void handleSingleKStreamOutbound(Map<String, ResolvableType> resolvableTypes, ResolvableType outboundResolvableType,
KStream<Object, Object> result, Iterator<String> outboundDefinitionIterator) {
if (outboundDefinitionIterator.hasNext()) {

60
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
View File

@@ -49,16 +49,16 @@ import org.springframework.util.CollectionUtils;
/**
* Kafka Streams specific target bindings proxy factory. See {@link AbstractBindableProxyFactory} for more details.
*
* <p>
* Targets bound by this factory:
*
* <p>
* {@link KStream}
* {@link KTable}
* {@link GlobalKTable}
*
* <p>
* This class looks at the Function bean's return signature as {@link ResolvableType} and introspect the individual types,
* binding them on the way.
*
* <p>
* 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<String> inputBindings = buildInputBindings();
Iterator<String> 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);

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java
View File

@@ -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

57
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
View File

@@ -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) {

47
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
View File

@@ -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<String, ResolvableType> resolvableTypeMap;
private final KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories;
private final Map<String, Method> methods;
private final StreamFunctionProperties streamFunctionProperties;
public KafkaStreamsFunctionProcessorInvoker(Map<String, ResolvableType> resolvableTypeMap,
KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor,
KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories,
Map<String, Method> methods) {
KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor,
KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories,
Map<String, Method> 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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> proxyFactory =
Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(functionUnit)).findFirst();
this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(functionUnit), functionUnit,
proxyFactory.get(), methods.get(functionUnit), null);
}
}
}
}

468
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java Normal file
View File

@@ -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<String, String> 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<String, Object> 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<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, String> 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<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("foo");
template.sendDefault("foobar!!");
template.setDefaultTopic("buzz");
template.sendDefault("this is crazy!!!");
Thread.sleep(1000);
ConsumerRecord<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<String, String> 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<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<String, String> 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<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("foo");
template.sendDefault("foo", "foobar!!");
Thread.sleep(1000);
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "bar");
assertThat(cr.value().contains("foobar!!")).isTrue();
}
finally {
pf.destroy();
}
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig1 {
@Bean
public Function<KStream<String, String>, KStream<String, String>> fooFunc() {
return input -> input.peek((s, s2) -> {
System.out.println("hello: " + s2);
});
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> anotherFooFunc() {
return input -> input.peek((s, s2) -> System.out.println("hello Foo: " + s2));
}
@Bean
public java.util.function.Consumer<KStream<String, String>> anotherProcess() {
return c -> c.foreach((s, s2) -> {
System.out.println("s2s2s2::" + s2);
countDownLatch1.countDown();
});
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig2 {
@Bean
public Function<KStream<String, String>, KStream<String, String>> fooFunc() {
return input -> input.peek((s, s2) -> {
System.out.println("hello: " + s2);
});
}
@Bean
public java.util.function.Consumer<KStream<String, String>> anotherProcess() {
return c -> c.foreach((s, s2) -> {
System.out.println("s2s2s2::" + s2);
countDownLatch2.countDown();
});
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig3 {
@Bean
public BiFunction<KStream<String, String>, KStream<String, String>, KStream<String, String>> fooBiFunc() {
return KStream::merge;
}
@Bean
public java.util.function.Consumer<KStream<String, String>> anotherProcess() {
return c -> c.foreach((s, s2) -> {
System.out.println("s2s2s2::" + s2);
countDownLatch3.countDown();
});
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig4 {
@Bean
public BiFunction<KStream<String, String>, KTable<String, String>, KStream<String, String>> fooBiFunc() {
return (a, b) -> a.join(b, (value1, value2) -> value1 + value2);
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> anotherFooFunc() {
return input -> input.mapValues(value -> value + "From-anotherFooFunc");
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> yetAnotherFooFunc() {
return input -> input.mapValues(value -> value + "From-yetAnotherFooFunc");
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> lastFunctionInChain() {
return input -> input.mapValues(value -> value + "From-lastFunctionInChain");
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig5 {
@Bean
public Function<KStream<String, String>, Function<KTable<String, String>, KStream<String, String>>> curriedFunc() {
return a -> b ->
a.join(b, (value1, value2) -> value1 + value2);
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> anotherFooFunc() {
return input -> input.mapValues(value -> value + "From-anotherFooFunc");
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> yetAnotherFooFunc() {
return input -> input.mapValues(value -> value + "From-yetAnotherFooFunc");
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> lastFunctionInChain() {
return input -> input.mapValues(value -> value + "From-lastFunctionInChain");
}
}
@EnableAutoConfiguration
public static class FunctionCompositionConfig6 {
@Bean
public Function<KStream<String, String>, KTable<String, String>> fooFunc() {
return ks -> {
ks.foreach(new ForeachAction<String, String>() {
@Override
public void apply(String key, String value) {
System.out.println();
}
});
return ks.toTable();
};
}
@Bean
public Function<KTable<String, String>, KStream<String, String>> anotherProcess() {
return KTable::toStream;
}
}
}