From 672f2bcebd6f3ba2885cce08bb68773b54d95190 Mon Sep 17 00:00:00 2001
From: Chris Bono <chris.bono@gmail.com>
Date: Tue, 24 May 2022 08:38:23 -0500
Subject: [PATCH] Discard raw/wildcard serde matches for KafkaStreams type
 inference (#2405)

---
 .../binder/kafka/streams/SerdeResolverUtils.java   | 14 +++++++++++---
 .../kafka/streams/SerdeResolverUtilsTests.java     |  9 +++++++++
 bom/spring-cloud-starter-parent/pom.xml            |  1 +
 3 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java
index 1f15480cc..fb6d6ed59 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java
@@ -141,8 +141,10 @@ abstract class SerdeResolverUtils {
 	 * @return list of bean names for matching serdes ordered by most specific match, or an empty list if no matches found
 	 */
 	static List<String> beanNamesForMatchingSerdes(ConfigurableApplicationContext context, ResolvableType targetType) {
-		// We don't attempt to find a matching Serde for type '?'
 		if (targetType.getRawClass() == null) {
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("Not looking for matching Serdes for raw or wildcard type: " + targetType);
+			}
 			return Collections.emptyList();
 		}
 		List<SerdeWithSpecificityScore> matchingSerdes = new ArrayList<>();
@@ -154,9 +156,15 @@ abstract class SerdeResolverUtils {
 				BeanDefinition beanDefinition = context.getBeanFactory().getMergedBeanDefinition(beanName);
 				ResolvableType serdeBeanGeneric = beanDefinition.getResolvableType().getGeneric(0);
 				if (LOG.isDebugEnabled()) {
-					LOG.debug("Found matching Serde<" + serdeBeanGeneric.getType() + "> under beanName=" + beanName);
+					LOG.debug("Found matching Serde<" + serdeBeanGeneric.getType() + "> with beanName=" + beanName);
+				}
+				Class<?> serdeGenericRawClazz = serdeBeanGeneric.getRawClass();
+				if (serdeGenericRawClazz != null) {
+					matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), beanName));
+				}
+				else if (LOG.isDebugEnabled()) {
+					LOG.debug("Discarding raw or wildcard match Serde<" + serdeBeanGeneric.getType() + "> with beanName=" + beanName);
 				}
-				matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), beanName));
 			}
 			catch (Exception ex) {
 				LOG.warn("Failed introspecting Serde bean '" + beanName + "'", ex);
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
index bfa4d2d77..c57bd9123 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
@@ -173,6 +173,15 @@ class SerdeResolverUtilsTests {
 				});
 		}
 
+		@Test
+		void discardsWildcardMatchesForJavaLangObject() {
+			new ApplicationContextRunner().withUserConfiguration(SerdeResolverSimpleTestApp.class)
+				.run((context) -> {
+					ResolvableType javaLangObjectSerdeType = ResolvableType.forClassWithGenerics(Serde.class, Object.class).getGeneric(0);
+					assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, javaLangObjectSerdeType)).isEmpty();
+				});
+		}
+
 		/**
 		 *
 		 * Verify that {@link SerdeResolverUtils#beanNamesForMatchingSerdes} returns the proper serdes in the proper order
diff --git a/bom/spring-cloud-starter-parent/pom.xml b/bom/spring-cloud-starter-parent/pom.xml
index 5083caf69..6bd9bd01b 100644
--- a/bom/spring-cloud-starter-parent/pom.xml
+++ b/bom/spring-cloud-starter-parent/pom.xml
@@ -7,6 +7,7 @@
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
 		<version>2.6.7</version>
+		<relativePath/>
 	</parent>
 	<groupId>org.springframework.cloud</groupId>
 	<artifactId>spring-cloud-stream-starter-parent</artifactId>