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

GH-2941: Kafka Streams binder composition issues

Browse Source 
* When composed functions are used on component functions in the Kafka Streams binder,
  there is an issue in which the first function in the composition is invoked twice.
  Fixing this issue by ensuring that the function execution path is only invoked once.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2941
This commit is contained in:
Soby Chacko
2024-05-29 11:54:05 -04:00
parent 625f25130d
commit 0e2954c01e
1 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -298,16 +298,20 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
}
else if (Function.class.isAssignableFrom(bean.getClass()) || BiFunction.class.isAssignableFrom(bean.getClass())) {
Object result;
if (BiFunction.class.isAssignableFrom(bean.getClass())) {
result = ((BiFunction) bean).apply(adaptedInboundArguments[0], adaptedInboundArguments[1]);
if (composedFunctionNames.length > 0) {
result = handleComposedFunctions(adaptedInboundArguments, null, composedFunctionNames);
}
else {
result = ((Function) bean).apply(adaptedInboundArguments[0]);
}
result = handleCurriedFunctions(adaptedInboundArguments, result);
if (composedFunctionNames.length > 0) {
result = handleComposedFunctions(adaptedInboundArguments, result, composedFunctionNames);
if (BiFunction.class.isAssignableFrom(bean.getClass())) {
result = ((BiFunction) bean).apply(adaptedInboundArguments[0], adaptedInboundArguments[1]);
}
else {
result = ((Function) bean).apply(adaptedInboundArguments[0]);
}
result = handleCurriedFunctions(adaptedInboundArguments, result);
}
if (result != null) {
final Set<String> outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs());
final Iterator<String> outboundDefinitionIterator = outputs.iterator();