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

KafkaStreams Functions Detection Logic (#2738)

Browse Source 
* KafkaStreams Functions Detection Logic

Kafka Streams functions declared in super classes are no longer
detected by the binder discovery algorithms. Fixing this issue
by properly scanning the methods from the super classes.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2737

* Addressing PR review
This commit is contained in:
Soby Chacko
2023-05-22 09:34:48 -04:00
committed by Soby Chacko
parent ee01cc9e50
commit 8d95e96df5
4 changed files with 43 additions and 11 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/KafkaStreamsBinderUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -16,9 +16,12 @@
package org.springframework.cloud.stream.binder.kafka.streams;
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;
import java.util.function.BiFunction;
import org.apache.commons.logging.Log;
@@ -64,7 +67,7 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Gary Russell
*/
final class KafkaStreamsBinderUtils {
public final class KafkaStreamsBinderUtils {
private static final Log LOGGER = LogFactory.getLog(KafkaStreamsBinderUtils.class);
@@ -72,6 +75,17 @@ final class KafkaStreamsBinderUtils {
}
/**
* Utility method to find the method targeted by the key.
*
* @param key name of the method
* @param methods collection of methods to search from
* @return found method as an {@link Optional}
*/
public static Optional<Method> findMethodWithName(String key, Method[] methods) {
return Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
}
static void prepareConsumerBinding(String name, String group,
ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner,
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,

14
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
@@ -37,6 +37,7 @@ import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderUtils;
import org.springframework.context.annotation.ConditionContext;
import org.springframework.core.ResolvableType;
import org.springframework.core.type.AnnotatedTypeMetadata;
@@ -94,12 +95,17 @@ public class FunctionDetectorCondition extends SpringBootCondition {
try {
Method[] methods = classObj.getDeclaredMethods();
Optional<Method> kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
// check if the bean name is overridden.
Optional<Method> kafkaStreamMethod = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
if (kafkaStreamMethod.isEmpty()) {
// check any inherited methods
methods = classObj.getMethods();
kafkaStreamMethod = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
}
if (kafkaStreamMethod.isEmpty()) {
// check if the bean name is overridden.
final BeanDefinition beanDefinition = context.getBeanFactory().getBeanDefinition(key);
final String factoryMethodName = beanDefinition.getFactoryMethodName();
kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst();
kafkaStreamMethod = KafkaStreamsBinderUtils.findMethodWithName(factoryMethodName, methods);
}
if (kafkaStreamMethod.isPresent()) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
@@ -48,6 +48,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderUtils;
import org.springframework.cloud.stream.function.StreamFunctionProperties;
import org.springframework.core.ResolvableType;
import org.springframework.util.ClassUtils;
@@ -176,11 +177,15 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
ClassUtils.getDefaultClassLoader());
try {
Method[] methods = classObj.getDeclaredMethods();
Optional<Method> functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
Optional<Method> functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
if (functionalBeanMethods.isEmpty()) {
methods = classObj.getMethods(); // check the inherited methods
functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
}
if (functionalBeanMethods.isEmpty()) {
final BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(key);
final String factoryMethodName = beanDefinition.getFactoryMethodName();
functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst();
functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(factoryMethodName, methods);
}
if (functionalBeanMethods.isPresent()) {

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2022 the original author or authors.
* Copyright 2018-2023 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.
@@ -176,7 +176,7 @@ class KafkaStreamsBinderBootstrapTest {
}
@SpringBootApplication
static class SimpleKafkaStreamsApplication {
static class SimpleKafkaStreamsApplication extends BaseConfig {
@Bean
public Consumer<KStream<Object, String>> input1() {
@@ -192,6 +192,11 @@ class KafkaStreamsBinderBootstrapTest {
};
}
}
// Testing the scenario reported by https://github.com/spring-cloud/spring-cloud-stream/issues/2737
static class BaseConfig {
@Bean
public Consumer<GlobalKTable<Object, String>> input3() {
return s -> {
@@ -199,4 +204,6 @@ class KafkaStreamsBinderBootstrapTest {
};
}
}
}