From f25dbff2b7fc0d0c742dd674a9e392057a34c86d Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 22 Mar 2021 18:55:20 -0400 Subject: [PATCH] Kafka Streams component bean scanning enhancements When functional beans are discovered from libraries in the classpath, it causes issues when Kafka Streams functions are scanned and bootstrapped. Binder expects the users to provide function definition property although the application does not directly include these functional beans or is aware of it. Fixing this issue by excluding non kafka streams function from scanning. Null check around adding micrometer listener to the StreamsBuilder. Addressing issues raised by the comments here: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1030#issuecomment-804039087 --- .../streams/StreamsBuilderFactoryManager.java | 4 +- ...KafkaStreamsFunctionBeanPostProcessor.java | 61 +++++++++++++++---- 2 files changed, 52 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index 7a15eae7d..c7238a8c3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -81,7 +81,9 @@ class StreamsBuilderFactoryManager implements SmartLifecycle { .getStreamsBuilderFactoryBeans(); int n = 0; for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) { - streamsBuilderFactoryBean.addListener(this.listener); + if (this.listener != null) { + streamsBuilderFactoryBean.addListener(this.listener); + } streamsBuilderFactoryBean.start(); this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean); } 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 797b891c8..542a6a0ad 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 @@ -18,6 +18,7 @@ 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.List; import java.util.Map; import java.util.Optional; @@ -65,6 +66,9 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, private final StreamFunctionProperties streamFunctionProperties; + private Map kafkaStreamsOnlyResolvableTypes = new HashMap<>(); + private Map kafakStreamsOnlyMethods = new HashMap<>(); + public KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) { this.streamFunctionProperties = streamFunctionProperties; } @@ -89,10 +93,14 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, Stream.concat(Stream.of(biFunctionNames), Stream.of(biConsumerNames))); final List collect = concat.collect(Collectors.toList()); collect.removeIf(s -> Arrays.stream(EXCLUDE_FUNCTIONS).anyMatch(t -> t.equals(s))); + onlySingleFunction = collect.size() == 1; collect.stream() .forEach(this::extractResolvableTypes); + kafkaStreamsOnlyResolvableTypes.keySet().forEach(k -> addResolvableTypeInfo(k, kafkaStreamsOnlyResolvableTypes.get(k))); + kafakStreamsOnlyMethods.keySet().forEach(k -> addResolvableTypeInfo(k, kafakStreamsOnlyMethods.get(k))); + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; for (String s : getResolvableTypes().keySet()) { @@ -115,15 +123,15 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, ClassUtils.getDefaultClassLoader()); try { Method[] methods = classObj.getMethods(); - Optional kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst(); - if (!kafkaStreamMethod.isPresent()) { + Optional functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst(); + if (!functionalBeanMethods.isPresent()) { final BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(key); final String factoryMethodName = beanDefinition.getFactoryMethodName(); - kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst(); + functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst(); } - if (kafkaStreamMethod.isPresent()) { - Method method = kafkaStreamMethod.get(); + if (functionalBeanMethods.isPresent()) { + Method method = functionalBeanMethods.get(); ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj); final Class rawClass = resolvableType.getGeneric(0).getRawClass(); if (rawClass == KStream.class || rawClass == KTable.class || rawClass == GlobalKTable.class) { @@ -131,7 +139,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, resolvableTypeMap.put(key, resolvableType); } else { - addResolvableTypeInfo(key, resolvableType); + discoverOnlyKafkaStreamsResolvableTypes(key, resolvableType); } } } @@ -149,7 +157,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, this.methods.put(key, method); } else { - addResolvableTypeInfo(key, resolvableType); + discoverOnlyKafkaStreamsResolvableTypesAndMethods(key, resolvableType, method); } } } @@ -161,13 +169,42 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, } private void addResolvableTypeInfo(String key, ResolvableType resolvableType) { - final String definition = streamFunctionProperties.getDefinition(); - if (definition == null) { - throw new IllegalStateException("Multiple functions found, but function definition property is not set."); - } - else if (definition.contains(key)) { + if (kafkaStreamsOnlyResolvableTypes.size() == 1) { resolvableTypeMap.put(key, resolvableType); } + else { + final String definition = streamFunctionProperties.getDefinition(); + if (definition == null) { + throw new IllegalStateException("Multiple functions found, but function definition property is not set."); + } + else if (definition.contains(key)) { + resolvableTypeMap.put(key, resolvableType); + } + } + } + + private void discoverOnlyKafkaStreamsResolvableTypes(String key, ResolvableType resolvableType) { + kafkaStreamsOnlyResolvableTypes.put(key, resolvableType); + } + + private void discoverOnlyKafkaStreamsResolvableTypesAndMethods(String key, ResolvableType resolvableType, Method method) { + kafkaStreamsOnlyResolvableTypes.put(key, resolvableType); + kafakStreamsOnlyMethods.put(key, method); + } + + private void addResolvableTypeInfo(String key, Method method) { + if (kafakStreamsOnlyMethods.size() == 1) { + this.methods.put(key, method); + } + else { + final String definition = streamFunctionProperties.getDefinition(); + if (definition == null) { + throw new IllegalStateException("Multiple functions found, but function definition property is not set."); + } + else if (definition.contains(key)) { + this.methods.put(key, method); + } + } } private boolean isKafkaStreamsTypeFound(Method method) {