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

GH-2846: Regression issue in Kafka Streams binder

Browse Source 
- When @Configuration or @Component style beans are used in Kafka Streams
   binder applications, there exists a regression that prevents the binder
   infrastructure to correctly detect the beans. Addressing this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2846
This commit is contained in:
Soby Chacko
2023-11-07 16:32:31 -05:00
parent b8e4885d21
commit 92324641ef
1 changed files with 52 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -56,9 +56,9 @@ import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.core.type.StandardClassMetadata;
import org.springframework.core.type.StandardMethodMetadata;
import org.springframework.lang.NonNull;
import org.springframework.util.ClassUtils;
/**
* @author Soby Chacko
@@ -186,30 +186,47 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(key);
ResolvableType resolvableType = null;
Class<?> rawClass = null;
Optional<Method> functionalBeanMethods;
try {
if (beanDefinition instanceof AnnotatedBeanDefinition annotatedBeanDefinition) {
StandardMethodMetadata factoryMethodMetadata = (StandardMethodMetadata) annotatedBeanDefinition.getFactoryMethodMetadata();
if (factoryMethodMetadata != null) {
if (annotatedBeanDefinition.getFactoryMethodMetadata() instanceof StandardMethodMetadata factoryMethodMetadata) {
Method introspectedMethod = factoryMethodMetadata.getIntrospectedMethod();
resolvableType = ResolvableType.forMethodReturnType(introspectedMethod);
rawClass = resolvableType.getGeneric(0).getRawClass();
}
else {
Class<?> introspectedClass = ((StandardClassMetadata) annotatedBeanDefinition.getMetadata()).getIntrospectedClass();
Method[] methods = introspectedClass.getDeclaredMethods();
final Class<?> classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
this.beanFactory.getBeanDefinition(key))
.getMetadata().getClassName(),
ClassUtils.getDefaultClassLoader());
Method[] methods = classObj.getDeclaredMethods();
functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
if (functionalBeanMethods.isEmpty()) {
methods = classObj.getMethods(); // check the inherited methods
functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(key, methods);
}
if (functionalBeanMethods.isEmpty()) {
final String factoryMethodName = beanDefinition.getFactoryMethodName();
functionalBeanMethods = KafkaStreamsBinderUtils.findMethodWithName(factoryMethodName, methods);
}
Optional<Method> componentBeanMethods = Arrays.stream(methods)
.filter(m -> m.getName().equals("apply") && isKafkaStreamsTypeFound(m) ||
m.getName().equals("accept") && isKafkaStreamsTypeFound(m)).findFirst();
if (componentBeanMethods.isPresent()) {
Method method = componentBeanMethods.get();
resolvableType = ResolvableType.forMethodParameter(method, 0);
rawClass = resolvableType.getRawClass();
if (onlySingleFunction) {
this.methods.put(key, method);
if (functionalBeanMethods.isPresent()) {
Method method = functionalBeanMethods.get();
resolvableType = ResolvableType.forMethodReturnType(method, classObj);
rawClass = resolvableType.getGeneric(0).getRawClass();
if (rawClass == KStream.class || rawClass == KTable.class || rawClass == GlobalKTable.class) {
saveTypeInformation(key, resolvableType);
}
else {
kafakStreamsOnlyMethods.put(key, method);
}
else {
Optional<Method> componentBeanMethods = Arrays.stream(methods)
.filter(m -> m.getName().equals("apply") && isKafkaStreamsTypeFound(m) ||
m.getName().equals("accept") && isKafkaStreamsTypeFound(m)).findFirst();
if (componentBeanMethods.isPresent()) {
Method method = componentBeanMethods.get();
resolvableType = ResolvableType.forMethodParameter(method, 0);
rawClass = resolvableType.getRawClass();
saveMethodInfoForComponentBeans(key, method);
}
}
}
@@ -220,12 +237,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
}
if (rawClass == KStream.class || rawClass == KTable.class || rawClass == GlobalKTable.class) {
if (onlySingleFunction) {
resolvableTypeMap.put(key, resolvableType);
}
else {
discoverOnlyKafkaStreamsResolvableTypes(key, resolvableType);
}
saveTypeInformation(key, resolvableType);
}
}
catch (Exception e) {
@@ -233,6 +245,24 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
}
}
private void saveMethodInfoForComponentBeans(String key, Method method) {
if (onlySingleFunction) {
this.methods.put(key, method);
}
else {
kafakStreamsOnlyMethods.put(key, method);
}
}
private void saveTypeInformation(String key, ResolvableType resolvableType) {
if (onlySingleFunction) {
resolvableTypeMap.put(key, resolvableType);
}
else {
discoverOnlyKafkaStreamsResolvableTypes(key, resolvableType);
}
}
private void addResolvableTypeInfo(String key, ResolvableType resolvableType) {
if (kafkaStreamsOnlyResolvableTypes.size() == 1) {
resolvableTypeMap.put(key, resolvableType);