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

GH-2817: Method name clash in Kafka Streams binder

Browse Source 
- When there are two methods with the same name but with different type erasures,
   Kafka Streams binder sometimes detects the incorrect method. Fixing this issue
   by specifically type checking the return type for Kafka Streams types.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2817
This commit is contained in:
Soby Chacko
2023-09-29 21:24:27 -04:00
parent d1a0aff623
commit cf6cea6525
2 changed files with 22 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java
View File

@@ -30,7 +30,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanDefinition;
@@ -49,6 +51,7 @@ import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolve
import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction;
import org.springframework.context.ApplicationContext;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaOperations;
@@ -84,7 +87,19 @@ public final class KafkaStreamsBinderUtils {
* @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();
return Arrays.stream(methods).filter(m -> m.getName().equals(key) &&
returnTypeContainsKafkaStreamsTypes(m)).findFirst();
}
private static boolean returnTypeContainsKafkaStreamsTypes(Method method) {
ResolvableType resolvableType = ResolvableType.forMethodReturnType(method);
ResolvableType[] generics = resolvableType.getGenerics();
if (generics.length > 0) {
Class<?> rawClass = generics[0].getRawClass();
return rawClass != null && (rawClass.isAssignableFrom(KStream.class) || rawClass.isAssignableFrom(KTable.class)
|| rawClass.isAssignableFrom(GlobalKTable.class));
}
return false;
}
public static String[] deriveFunctionUnits(String definition) {

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.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.
@@ -222,6 +222,11 @@ public class MultipleFunctionsInSameAppTests {
(s, p) -> p.equalsIgnoreCase("electronics"));
}
// Testing for the scenario under https://github.com/spring-cloud/spring-cloud-stream/issues/2817
public String processItem(String foo) {
return "testing";
}
@Bean
public Function<KStream<String, String>, KStream<String, Long>> yetAnotherProcess() {
return input -> input.map((k, v) -> new KeyValue<>("foo", 1L));