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

Cleanup SerdeResolverUtils

Browse Source 
* Refactor SerdeResolverUtils logic
* Add more tests in SerdeResolverUtilsTests
* Fixed some random test warnings

Fixes #2370

Code review feedback: remove FQ ref to inner method

Fix checkstyles
This commit is contained in:
Chris Bono
2022-05-12 18:46:41 -05:00
committed by Soby Chacko
parent a827177ac1
commit 19b867cbf5
3 changed files with 523 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -207,7 +207,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
* @return {@link Serde} for the state store key.
*/
public Serde<?> getStateStoreKeySerde(String keySerdeString) {
return getKeySerde(keySerdeString, (Map<String, ?>) null);
return getKeySerde(keySerdeString, null);
}
/**
@@ -217,7 +217,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
*/
public Serde<?> getStateStoreValueSerde(String valueSerdeString) {
try {
return getValueSerde(valueSerdeString, (Map<String, ?>) null);
return getValueSerde(valueSerdeString, null);
}
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
@@ -242,20 +242,18 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
}
private Serde<?> getKeySerde(String keySerdeString, ResolvableType resolvableType, Map<String, ?> extendedConfiguration) {
Serde<?> keySerde = null;
Serde<?> keySerde;
try {
if (StringUtils.hasText(keySerdeString)) {
keySerde = Utils.newInstance(keySerdeString, Serde.class);
}
else {
keySerde = Serdes.ByteArray();
if (resolvableType != null &&
(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
Serde<?> fallbackSerde = getFallbackSerde("default.key.serde");
keySerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
}
if (keySerde == null) {
keySerde = Serdes.ByteArray();
keySerde = SerdeResolverUtils.resolveForType(this.context, targetType, fallbackSerde);
}
}
keySerde.configure(combineStreamConfigProperties(extendedConfiguration), true);
@@ -298,24 +296,19 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
: Serdes.ByteArray();
}
@SuppressWarnings("unchecked")
private Serde<?> getValueSerde(String valueSerdeString, ResolvableType resolvableType, Map<String, ?> extendedConfiguration)
throws ClassNotFoundException {
Serde<?> valueSerde = null;
Serde<?> valueSerde;
if (StringUtils.hasText(valueSerdeString)) {
valueSerde = Utils.newInstance(valueSerdeString, Serde.class);
}
else {
valueSerde = Serdes.ByteArray();
if (resolvableType != null && ((isResolvalbeKafkaStreamsType(resolvableType)) ||
(isResolvableKStreamArrayType(resolvableType)))) {
ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
Serde<?> fallbackSerde = getFallbackSerde("default.value.serde");
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
valueSerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
}
if (valueSerde == null) {
valueSerde = Serdes.ByteArray();
valueSerde = SerdeResolverUtils.resolveForType(this.context, targetType, fallbackSerde);
}
}
valueSerde.configure(combineStreamConfigProperties(extendedConfiguration), false);
@@ -329,7 +322,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
private Map<String, ?> combineStreamConfigProperties(Map<String, ?> extendedConfiguration) {
if (extendedConfiguration != null && !extendedConfiguration.isEmpty()) {
Map<String, Object> streamConfiguration = new HashMap(this.streamConfigGlobalProperties);
Map<String, Object> streamConfiguration = new HashMap<>(this.streamConfigGlobalProperties);
streamConfiguration.putAll(extendedConfiguration);
return streamConfiguration;
}

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

@@ -16,15 +16,13 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
@@ -32,11 +30,12 @@ import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.util.ClassUtils;
import org.springframework.lang.Nullable;
/**
* Utility class that contains various methods to help resolve {@link Serde Serdes}.
@@ -48,124 +47,120 @@ abstract class SerdeResolverUtils {
private static final Log LOG = LogFactory.getLog(SerdeResolverUtils.class);
/** Classnames of the standard built-in Serdes supported in {@link Serdes#serdeFrom(Class)}. */
private static final Set<String> STANDARD_SERDE_CLASSNAMES = Set.of(
Serdes.String().getClass().getName(),
Serdes.Short().getClass().getName(),
Serdes.Integer().getClass().getName(),
Serdes.Long().getClass().getName(),
Serdes.Float().getClass().getName(),
Serdes.Double().getClass().getName(),
Serdes.ByteArray().getClass().getName(),
Serdes.ByteBuffer().getClass().getName(),
Serdes.Bytes().getClass().getName(),
Serdes.UUID().getClass().getName());
/**
* Return the closest matching configured {@code Serde<?>} bean if one exists, or the specified
* {@code fallbackSerde}, or finally a standard default serde if no fallback specified.
*
* Return the {@code Serde<?>} to use for the specified type using the following steps until a match is found.
* <p><ul>
* <li>the closest matching configured {@code Serde<?>} bean if one exists</li>
* <li>the Kafka Streams built-in serde if the target type is one of the built-in types exposed by Kafka Streams
* (Integer, Long, Short, Double, Float, byte[], UUID and String)
* <li>the fallback serde if specified and not one of the Kafka Streams exposed type serdes</li>
* <li>the {@link JsonSerde} if the target type is not exactly {@code Object}</li>
* <li>the fallback as the last resort</li>
* </ul>
* @param context the application context
* @param targetType the target type to find the serde for
* @param fallbackSerde the fallback serde in case no matching serde bean found in the context
* @return serde to use for the target type
* @param fallbackSerde the serde to use when no type can be inferred
* @return serde to use for the target type or {@code fallbackSerde} as outlined in the method description
*/
static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, Serde<?> fallbackSerde) {
static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, @Nullable Serde<?> fallbackSerde) {
List<Serde<?>> matchingSerdes = findMatchingSerdes(context, targetType);
if (!matchingSerdes.isEmpty()) {
return matchingSerdes.get(0);
}
// We don't attempt to find a matching Serde for type '?'
if (targetType.getRawClass() == null) {
return null;
}
Serde<?> serde = null;
Class<?> genericRawClazz = targetType.getRawClass();
if (Integer.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Integer();
// We don't attempt to find a matching Serde for type '?' - just return fallback
if (genericRawClazz == null) {
return fallbackSerde;
}
else if (Long.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Long();
List<String> matchingSerdes = beanNamesForMatchingSerdes(context, targetType);
if (!matchingSerdes.isEmpty()) {
return context.getBean(matchingSerdes.get(0), Serde.class);
}
else if (Short.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Short();
// Use standard serde for built-in types
Serde<?> standardDefaultSerde = getStandardDefaultSerde(genericRawClazz);
if (standardDefaultSerde != null) {
return standardDefaultSerde;
}
else if (Double.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Double();
// Use fallback if specified and not from std defaults (we know from above that type is not std default
// so using a fallback that is std default type would not work)
if (fallbackSerde != null && !isSerdeFromStandardDefaults(fallbackSerde)) {
return fallbackSerde;
}
else if (Float.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Float();
// Use JsonSerde if type is not exactly Object
if (!genericRawClazz.isAssignableFrom((Object.class))) {
return new JsonSerde<>(genericRawClazz);
}
else if (byte[].class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.ByteArray();
// Finally, just resort to using the fallback
return fallbackSerde;
}
private static Serde<?> getStandardDefaultSerde(Class<?> genericRawClazz) {
try {
return Serdes.serdeFrom(genericRawClazz);
}
else if (String.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.String();
}
else if (UUID.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.UUID();
}
else if (!isSerdeFromStandardDefaults(fallbackSerde)) {
// User purposely set a default serde that is not one of the above
serde = fallbackSerde;
}
else {
// If the type is Object, then skip assigning the JsonSerde and let the fallback mechanism takes precedence.
if (!genericRawClazz.isAssignableFrom((Object.class))) {
serde = new JsonSerde(genericRawClazz);
catch (IllegalArgumentException ex) {
if (LOG.isTraceEnabled()) {
LOG.trace(ex);
}
}
return serde;
return null;
}
private static boolean isSerdeFromStandardDefaults(Serde<?> serde) {
if (serde != null) {
if (Number.class.isAssignableFrom(serde.getClass())) {
return true;
}
else if (Serdes.ByteArray().getClass().isAssignableFrom(serde.getClass())) {
return true;
}
else if (Serdes.String().getClass().isAssignableFrom(serde.getClass())) {
return true;
}
else if (Serdes.UUID().getClass().isAssignableFrom(serde.getClass())) {
return true;
}
if (serde == null) {
return false;
}
return false;
return STANDARD_SERDE_CLASSNAMES.contains(serde.getClass().getName());
}
/**
* Find all {@link Serde} beans that are assignable from {@code targetType}.
* Find the names of all {@link Serde} beans that can be used for {@code targetType}.
*
* @param context the application context
* @param targetType the target type the serdes are being matched for
* @return list of matching serdes order by most specific match, or an empty list if no matches found
* @return list of bean names for matching serdes ordered by most specific match, or an empty list if no matches found
*/
static List<Serde<?>> findMatchingSerdes(ConfigurableApplicationContext context, ResolvableType targetType) {
static List<String> beanNamesForMatchingSerdes(ConfigurableApplicationContext context, ResolvableType targetType) {
// We don't attempt to find a matching Serde for type '?'
if (targetType.getRawClass() == null) {
return Collections.emptyList();
}
List<SerdeWithSpecificityScore> matchingSerdes = new ArrayList<>();
context.getBeansOfType(Serde.class).forEach((beanName, serdeBean) -> {
final Class<?> beanConfigClass = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
context.getBeanFactory().getBeanDefinition(beanName))
.getMetadata().getClassName(),
ClassUtils.getDefaultClassLoader());
ResolvableType serdeType = ResolvableType.forClassWithGenerics(Serde.class, targetType);
String[] serdeBeanNames = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context.getBeanFactory(),
serdeType, false, false);
Arrays.stream(serdeBeanNames).forEach((beanName) -> {
try {
Method[] methods = beanConfigClass.getMethods();
Optional<Method> serdeBeanMethod = Arrays.stream(methods).filter(m -> m.getName().equals(beanName)).findFirst();
serdeBeanMethod.ifPresent((method) -> {
ResolvableType serdeBeanMethodReturnType = ResolvableType.forMethodReturnType(method, beanConfigClass);
ResolvableType serdeBeanGeneric = serdeBeanMethodReturnType.getGeneric(0);
// We don't attempt to use a Serde<?> as a match for anything currently
if (serdeBeanGeneric.getRawClass() != null && serdeBeanGeneric.isAssignableFrom(targetType)) {
matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), serdeBean));
}
});
}
catch (Exception e) {
if (LOG.isTraceEnabled()) {
LOG.trace("Failed to introspect Serde bean method '" + serdeBean + "'", e);
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);
}
matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), beanName));
}
catch (Exception ex) {
LOG.warn("Failed introspecting Serde bean '" + beanName + "'", ex);
}
});
if (!matchingSerdes.isEmpty()) {
return matchingSerdes.stream().sorted(Collections.reverseOrder())
.map(SerdeWithSpecificityScore::getSerde)
.map(SerdeWithSpecificityScore::getSerdeBeanName)
.collect(Collectors.toList());
}
return Collections.emptyList();
@@ -182,7 +177,7 @@ abstract class SerdeResolverUtils {
* <p><br><b>Example:</b>
* <pre>{@code
* -------------------------------------------------------------------------------------------------------
* targetType: Foo<Date> toString='Foo<Date>' typeName='Foo<java.util.Date>'
* targetType: Foo<Date> toString='Foo<Date>' typeName='Foo<java.util.Date>'
* typeToCheck1: Foo<Date> toString='Foo<Date>' typeName='Foo<java.util.Date>'
* typeToCheck2: Foo<? extends Date> toString='Foo<Date>' typeName='Foo<? extends java.util.Date>'
* -------------------------------------------------------------------------------------------------------
@@ -221,15 +216,15 @@ abstract class SerdeResolverUtils {
*/
private static class SerdeWithSpecificityScore implements Comparable<SerdeWithSpecificityScore> {
private Integer score;
private Serde<?> serde;
private String serdeBeanName;
SerdeWithSpecificityScore(Integer score, Serde<?> serde) {
SerdeWithSpecificityScore(Integer score, String serdeBeanName) {
this.score = Objects.requireNonNull(score);
this.serde = Objects.requireNonNull(serde);
this.serdeBeanName = Objects.requireNonNull(serdeBeanName);
}
Serde<?> getSerde() {
return serde;
String getSerdeBeanName() {
return serdeBeanName;
}
@Override

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

@@ -16,77 +16,349 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.UUID;
import java.util.stream.Stream;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.Bytes;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.support.serializer.JsonSerde;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.params.provider.Arguments.arguments;
import static org.mockito.Mockito.mock;
/**
* Unit tests for {@link SerdeResolverUtils}.
*
* @author Chris Bono
*/
@SuppressWarnings({ "rawtypes", "NewClassNamingConvention", "unchecked" })
class SerdeResolverUtilsTests {
/**
* Verify that {@link SerdeResolverUtils#findMatchingSerdes} returns the proper serdes
* in the proper order for the following grid:
* <p><br>
* <pre>{@code
* ------------------------------------------------------------------
* KStream type | Serde type
* ------------------------------------------------------------------
* | GE<Date> | GE<? extends Date> | GE<?> | GE
* ------------------------------------------------------------------
* GE<Date> | Y | Y | Y | Y
* GE<? extends Date> | N | Y | Y | N
* GE<?> | N | N | Y | N
* GE | N | N | N | N
* ------------------------------------------------------------------
* }</pre>
*/
@Test
void findMatchingSerdesForSimpleGenericType() {
@Nested
class ResolveForType {
ResolvableType geDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Date>>() { });
ResolvableType geBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Date>>() { });
ResolvableType geWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<?>>() { });
ResolvableType geRaw = ResolvableType.forRawClass(GenericEvent.class);
private ApplicationContextRunner contextRunner = new ApplicationContextRunner();
ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(SerdeResolverTestApp.class);
private Serde<?> fallback = mock(Serde.class);
contextRunner.run((context) -> {
@Test
void returnsFallbackSerdeForWildcard() {
this.contextRunner
.withConfiguration(AutoConfigurations.of(SerdeResolverSimpleTestApp.class))
.run((context) -> {
ResolvableType wildcardType = ResolvableType.forClass(Serde.class).getGeneric(0);
assertThat(SerdeResolverUtils.resolveForType(context, wildcardType, fallback)).isSameAs(fallback);
});
}
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geDate))
.extracting("name")
.containsExactly("genericEventDateSerde", "genericEventDateBoundedSerde", "genericEventWildcardSerde", "genericEventRawSerde");
@Test
void returnsSerdeBeanForMatchingType() {
this.contextRunner
.withConfiguration(AutoConfigurations.of(SerdeResolverSimpleTestApp.class))
.run((context) -> {
ResolvableType fooType = ResolvableType.forClass(Foo.class);
assertThat(SerdeResolverUtils.resolveForType(context, fooType, fallback)).isInstanceOf(FooSerde.class);
});
}
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geBounded))
.extracting("name")
.containsExactly("genericEventDateBoundedSerde", "genericEventWildcardSerde");
@Nested
class NoMatchingSerdeBeans {
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geWildcard))
.extracting("name")
.containsExactly("genericEventWildcardSerde");
@ParameterizedTest
@MethodSource("kafkaStreamsBuiltInTypes")
void returnsStandardSerdeForKafkaStreamsBuiltInType(Class<?> builtInType, Serde<?> expectedBuiltInSerde) {
contextRunner.run((context) ->
assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(builtInType), fallback))
.isInstanceOf(expectedBuiltInSerde.getClass()));
}
// Because GenericEvent is a parameterized type, Serde<GenericEvent> resolves to Serde<GenericEvent<?>>
// which is not assignable from GenericEvent
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geRaw))
.extracting("name")
.isEmpty();
});
static Stream<Arguments> kafkaStreamsBuiltInTypes() {
return Stream.of(
arguments(String.class, Serdes.String()),
arguments(Short.class, Serdes.Short()),
arguments(Integer.class, Serdes.Integer()),
arguments(Long.class, Serdes.Long()),
arguments(Float.class, Serdes.Float()),
arguments(Double.class, Serdes.Double()),
arguments(byte[].class, Serdes.ByteArray()),
arguments(ByteBuffer.class, Serdes.ByteBuffer()),
arguments(Bytes.class, Serdes.Bytes()),
arguments(UUID.class, Serdes.UUID())
);
}
@Nested
class ForNonKafkaStreamsBuiltInType {
@Test
void returnsFallbackSerdeWhenValidFallbackSpecified() {
contextRunner.run((context) ->
assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), fallback))
.isSameAs(fallback));
}
@ParameterizedTest
@MethodSource("invalidFallbackSerdeProvider")
void returnsJsonSerdeWhenInvalidFallbackSpecified(Serde<?> invalidFallback) {
contextRunner.run((context) ->
assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), invalidFallback))
.isInstanceOf(JsonSerde.class));
}
static Stream<Serde<?>> invalidFallbackSerdeProvider() {
return Stream.of(
Serdes.String(),
Serdes.Short(),
Serdes.Integer(),
Serdes.Long(),
Serdes.Float(),
Serdes.Double(),
Serdes.ByteArray(),
Serdes.ByteBuffer(),
Serdes.Bytes(),
Serdes.UUID()
);
}
@Test
void returnsJsonSerdeWhenFallbackNotSpecified() {
contextRunner.run((context ->
assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), null))
.isInstanceOf(JsonSerde.class)));
}
@Test
void returnsFallbackSerdeForJavaLangObject() {
// This is an edge case as the only way to get to the JsonSerde step in the 1st place is when
// no fallback is specified or a fallback is specified but its invalid (for a built-in type).
// We will use the 'fallback is not specified' scenario for this test.
contextRunner.run((context) ->
assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Object.class), null))
.isNull());
}
}
}
}
@Nested
class BeanNamesForMatchingSerdes {
@Test
void returnsNoSerdesForWildcardType() {
new ApplicationContextRunner().withUserConfiguration(SerdeResolverSimpleTestApp.class)
.run((context) -> {
ResolvableType wildcardType = ResolvableType.forClass(Serde.class).getGeneric(0);
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, wildcardType)).isEmpty();
});
}
/**
*
* Verify that {@link SerdeResolverUtils#beanNamesForMatchingSerdes} returns the proper serdes in the proper order
* for the following grid:
* <p><br>
* <i>NOTE:</i> {@code GE = GenericEvent}
* <p><pre>{@code
* ------------------------------------------------------------------
* KStream type | Serde type
* ------------------------------------------------------------------
* | GE<Date> | GE<? extends Date> | GE<?> | GE
* ------------------------------------------------------------------
* GE<Date> | 1 | - | - | -
* GE<? extends Date> | 2 | 1 | - | -
* GE<?> | 4 | 2 | 1 | 3
* GE | 1 | - | - | 2
* ------------------------------------------------------------------
* }</pre>
* <p><br>
* <i>NOTE:</i> On the last row, one might expect the {@code GE} serde to be the top match with the {@code GE}
* kstream. However, that is not the case because {@code GE} is a parameterized type and therefore when
* specified as a raw type (without any type info) it resolves to {@code GE<?>} which throws off the ordering.
* A best practice is to specify the type info (even if it is wildcard) for KStream parameterized types.
*/
@Test
void returnsProperlyOrderedSerdesForSimpleGenericTypes() {
ResolvableType geDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Date>>() { });
ResolvableType geBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Date>>() { });
ResolvableType geWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<?>>() { });
ResolvableType geRaw = ResolvableType.forRawClass(GenericEvent.class);
new ApplicationContextRunner().withUserConfiguration(SerdeResolverSimpleTestApp.class).run((context) -> {
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geDate))
.containsExactly(
"geDateSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geBounded))
.containsExactly(
"geDateBoundedSerde",
"geDateSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geWildcard))
.containsExactly(
"geWildcardSerde",
"geDateBoundedSerde",
"geRawSerde",
"geDateSerde",
"geStringSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geRaw))
.containsExactly(
"geDateSerde",
"geStringSerde",
"geRawSerde");
});
}
/**
* Verify that {@link SerdeResolverUtils#beanNamesForMatchingSerdes} returns the proper serdes in the proper order
* for the following grid:
* <p><br>
* <i>NOTE:</i> {@code GE = GenericEvent}
* <p><pre>{@code
* -------------------------------------------------------------------------------------------------------------------
* KStream type | Serde type
* -------------------------------------------------------------------------------------------------------------------
* | GE<F<D>> | GE<F<?D>> | GE<F<?>> | GE<F> | GE<?F<D>> | GE<?F<?D>> | GE<?F<?>> | GE<?F> | GE<?> | GE
* -------------------------------------------------------------------------------------------------------------------
* GE<F<D>> | 1 | - | - | - | - | - | - | - | - | -
* GE<F<?D>> | 2 | 1 | - | - | - | - | - | - | - | -
* GE<F<?>> | 4 | 3 | 1 | 2 | - | - | - | - | - | -
* GE<F> | 2 | - | - | 1 | - | - | - | - | - | -
* GE<?F<D>> | 2 | - | - | - | 1 | - | - | - | - | -
* GE<?F<?D>> | 3 | 4 | - | - | 2 | 1 | - | - | - | -
* GE<?F<?>> | 7 | 8 | 5 | 6 | 4 | 3 | 1 | 2 | - | -
* GE<?F> | 4 | - | - | 3 | 2 | - | - | 1 | - | -
* GE<?> | 7 | 8 | 9 | 10 | 2 | 3 | 4 | 5 | 1 | 6
* GE | 1 | 2 | 3 | 4 | - | - | - | - | - | 5
* -------------------------------------------------------------------------------------------------------------------
* }</pre>
* <p><br>
* <i>NOTE:</i> On the last row, one might expect the {@code GE} serde to be the top match with the {@code GE}
* kstream. However, that is not the case because {@code GE} is a parameterized type and therefore when
* specified as a raw type (without any type info) it resolves to {@code GE<?>} which throws off the ordering.
* A best practice is to specify the type info (even if it is wildcard) for KStream parameterized types.
*/
@Test
void returnsProperlyOrderedSerdesForComplexGenericTypes() {
ResolvableType geFooDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<Date>>>() { });
ResolvableType geFooDateBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<? extends Date>>>() { });
ResolvableType geFooWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<?>>>() { });
ResolvableType geFooRaw = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo>>() { });
ResolvableType geFooBoundedDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<Date>>>() { });
ResolvableType geFooBoundedDateBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<? extends Date>>>() { });
ResolvableType geFooBoundedWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<?>>>() { });
ResolvableType geFooBoundedRaw = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo>>() { });
ResolvableType geWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<?>>() { });
ResolvableType geRaw = ResolvableType.forRawClass(GenericEvent.class);
new ApplicationContextRunner().withUserConfiguration(SerdeResolverComplexTestApp.class).run((context) -> {
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooDate))
.containsExactly(
"geFooDateSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooDateBounded))
.containsExactly(
"geFooDateBoundedSerde",
"geFooDateSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooWildcard))
.containsExactly(
"geFooWildcardSerde",
"geFooRawSerde",
"geFooDateBoundedSerde",
"geFooDateSerde",
"geFooStringSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooRaw))
.containsExactly(
"geFooRawSerde",
"geFooDateSerde",
"geFooStringSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedDate))
.containsExactly(
"geFooBoundedDateSerde",
"geFooDateSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedDateBounded))
.containsExactly(
"geFooBoundedDateBoundedSerde",
"geFooBoundedDateSerde",
"geFooDateSerde",
"geFooDateBoundedSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedWildcard))
.containsExactly(
"geFooBoundedWildcardSerde",
"geFooBoundedRawSerde",
"geFooBoundedDateBoundedSerde",
"geFooBoundedDateSerde",
"geFooBoundedStringSerde",
"geFooWildcardSerde",
"geFooRawSerde",
"geFooDateSerde",
"geFooDateBoundedSerde",
"geFooStringSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedRaw))
.containsExactly(
"geFooBoundedRawSerde",
"geFooBoundedDateSerde",
"geFooBoundedStringSerde",
"geFooRawSerde",
"geFooDateSerde",
"geFooStringSerde");
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geWildcard))
.containsExactly(
"geWildcardSerde",
"geFooBoundedDateSerde",
"geFooBoundedDateBoundedSerde",
"geFooBoundedStringSerde",
"geFooBoundedWildcardSerde",
"geFooBoundedRawSerde",
"geRawSerde",
"geFooDateSerde",
"geFooDateBoundedSerde",
"geFooStringSerde",
"geFooWildcardSerde",
"geFooRawSerde");
// One might expect geRawSerde to win in order, but it does not because GE is a parameterized type and
// therefore GE resolves to GE<?> which throws off the ordering. Bottom line, for parameterized types
// be sure to specify a type (even if it's wildcard) in the KStream<T> definition.
assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geRaw))
.containsExactly(
"geFooDateSerde",
"geFooDateBoundedSerde",
"geFooStringSerde",
"geFooWildcardSerde",
"geFooRawSerde",
"geRawSerde");
});
}
}
static class GenericEventSerde<T> implements Serde<GenericEvent<? extends T>> {
@@ -96,10 +368,6 @@ class SerdeResolverUtilsTests {
this.name = name;
}
String getName() {
return name;
}
@Override
public Serializer<GenericEvent<? extends T>> serializer() {
return null;
@@ -109,41 +377,141 @@ class SerdeResolverUtilsTests {
public Deserializer<GenericEvent<? extends T>> deserializer() {
return null;
}
@Override
public String toString() {
return "GenericEventSerde(" + name + ")";
}
}
static class GenericEvent<T> { }
static class FooSerde implements Serde<Foo> {
@Override
public Serializer<Foo> serializer() {
return null;
}
@Override
public Deserializer<Foo> deserializer() {
return null;
}
}
static class Foo<T> { }
@EnableAutoConfiguration
static class SerdeResolverTestApp {
static class SerdeResolverSimpleTestApp {
@Bean
public Serde<GenericEvent<Date>> genericEventDateSerde() {
return new GenericEventSerde("genericEventDateSerde");
public Serde<GenericEvent<Date>> geDateSerde() {
return new GenericEventSerde("geDateSerde");
}
@Bean
public Serde<GenericEvent<? extends Date>> genericEventDateBoundedSerde() {
return new GenericEventSerde("genericEventDateBoundedSerde");
public Serde<GenericEvent<? extends Date>> geDateBoundedSerde() {
return new GenericEventSerde("geDateBoundedSerde");
}
@Bean
public Serde<GenericEvent<String>> genericEventStringSerde() {
return new GenericEventSerde("genericEventStringSerde");
public Serde<GenericEvent<String>> geStringSerde() {
return new GenericEventSerde("geStringSerde");
}
@Bean
public Serde<GenericEvent<?>> genericEventWildcardSerde() {
return new GenericEventSerde("genericEventWildcardSerde");
public Serde<GenericEvent<?>> geWildcardSerde() {
return new GenericEventSerde("geWildcardSerde");
}
@Bean
public Serde<GenericEvent> genericEventRawSerde() {
return new GenericEventSerde("genericEventRawSerde");
public Serde<GenericEvent> geRawSerde() {
return new GenericEventSerde("geRawSerde");
}
@Bean
public Serde<?> widlcardSerde() {
return Serdes.Void();
}
@Bean
public Serde<Foo> fooSerde() {
return new FooSerde();
}
}
@EnableAutoConfiguration
static class SerdeResolverComplexTestApp {
@Bean
public Serde<GenericEvent<Foo<Date>>> geFooDateSerde() {
return new GenericEventSerde("geFooDateSerde");
}
@Bean
public Serde<GenericEvent<Foo<? extends Date>>> geFooDateBoundedSerde() {
return new GenericEventSerde("geFooDateBoundedSerde");
}
@Bean
public Serde<GenericEvent<Foo<String>>> geFooStringSerde() {
return new GenericEventSerde("geFooStringSerde");
}
@Bean
public Serde<GenericEvent<Foo<?>>> geFooWildcardSerde() {
return new GenericEventSerde("geFooWildcardSerde");
}
@Bean
public Serde<GenericEvent<Foo>> geFooRawSerde() {
return new GenericEventSerde("geFooRawSerde");
}
@Bean
public Serde<GenericEvent<? extends Foo<Date>>> geFooBoundedDateSerde() {
return new GenericEventSerde("geFooBoundedDateSerde");
}
@Bean
public Serde<GenericEvent<? extends Foo<? extends Date>>> geFooBoundedDateBoundedSerde() {
return new GenericEventSerde("geFooBoundedDateBoundedSerde");
}
@Bean
public Serde<GenericEvent<? extends Foo<String>>> geFooBoundedStringSerde() {
return new GenericEventSerde("geFooBoundedStringSerde");
}
@Bean
public Serde<GenericEvent<? extends Foo<?>>> geFooBoundedWildcardSerde() {
return new GenericEventSerde("geFooBoundedWildcardSerde");
}
@Bean
public Serde<GenericEvent<? extends Foo>> geFooBoundedRawSerde() {
return new GenericEventSerde("geFooBoundedRawSerde");
}
@Bean
public Serde<GenericEvent<?>> geWildcardSerde() {
return new GenericEventSerde("geWildcardSerde");
}
@Bean
public Serde<GenericEvent> geRawSerde() {
return new GenericEventSerde("geRawSerde");
}
@Bean
public Serde<?> widlcardSerde() {
return Serdes.Void();
}
@Bean
public Serde<Foo> fooSerde() {
return new FooSerde();
}
}
}