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

GH-1156: Kafka Streams binder composition issues

Browse Source 
When both regular Kafka and Kafka Streams functions are present,
the code that was added recently for function composition in
Kafka Streams binder was accidentally creating a binadable proxy
factory bean for non Kafka Streams functions. Resolving this issue.

Resovles https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1156
This commit is contained in:
Soby Chacko
2021-09-28 15:41:01 -04:00
parent 9fd16416d6
commit bd1b49222c
2 changed files with 29 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,9 +19,11 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -107,6 +109,9 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
final String definition = streamFunctionProperties.getDefinition();
final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{};
final Set<String> kafkaStreamsMethodNames = new HashSet<>(kafkaStreamsOnlyResolvableTypes.keySet());
kafkaStreamsMethodNames.addAll(this.resolvableTypeMap.keySet());
if (functionUnits.length == 0) {
for (String s : getResolvableTypes().keySet()) {
ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(s)};
@@ -123,21 +128,30 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
ResolvableType[] resolvableTypes = new ResolvableType[composedFunctions.length];
int i = 0;
boolean nonKafkaStreamsFunctionsFound = false;
for (String split : composedFunctions) {
derivedNameFromComposed = derivedNameFromComposed.concat(split);
resolvableTypes[i++] = getResolvableTypes().get(split);
if (!kafkaStreamsMethodNames.contains(split)) {
nonKafkaStreamsFunctionsFound = true;
break;
}
}
if (!nonKafkaStreamsFunctionsFound) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
KafkaStreamsBindableProxyFactory.class);
registerKakaStreamsProxyFactory(registry, derivedNameFromComposed, resolvableTypes, rootBeanDefinition);
}
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);
// Ensure that the function unit is a Kafka Streams function
if (kafkaStreamsMethodNames.contains(functionUnit)) {
ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(functionUnit)};
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
KafkaStreamsBindableProxyFactory.class);
registerKakaStreamsProxyFactory(registry, functionUnit, resolvableTypes, rootBeanDefinition);
}
}
}
}

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

@@ -73,15 +73,16 @@ public class KafkaStreamsFunctionProcessorInvoker {
}
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);
proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory ->
this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]),
derivedNameFromComposed[0], kafkaStreamsBindableProxyFactory, 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);
proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory ->
this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(functionUnit), functionUnit,
kafkaStreamsBindableProxyFactory, methods.get(functionUnit), null));
}
}
}