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

Discard raw/wildcard serde matches for KafkaStreams type inference (#2405)

Browse Source
This commit is contained in:
Chris Bono
2022-05-24 08:38:23 -05:00
committed by GitHub
parent 483ffb68a7
commit 672f2bcebd
3 changed files with 21 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
View File

@@ -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

1
bom/spring-cloud-starter-parent/pom.xml
View File

@@ -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>