From bd1b49222cf9c6e463b165847571e84d77e24f7f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 28 Sep 2021 15:41:01 -0400 Subject: [PATCH] GH-1156: Kafka Streams binder composition issues 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 --- ...KafkaStreamsFunctionBeanPostProcessor.java | 32 +++++++++++++------ .../KafkaStreamsFunctionProcessorInvoker.java | 11 ++++--- 2 files changed, 29 insertions(+), 14 deletions(-) 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 71cf073fd..0e7e99785 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 @@ -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 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); + } } } } 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 acc2d9173..aa26f82e8 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 @@ -73,15 +73,16 @@ public class KafkaStreamsFunctionProcessorInvoker { } 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); + 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 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)); } } }