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

Consider generics when resolving Serdes for KafkaStreams binder (backport)

Browse Source 
Fixes #2318
This commit is contained in:
Chris Bono
2022-05-01 23:02:04 -05:00
parent 3ecbfc6c59
commit 0c698f921e
4 changed files with 528 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,15 +16,9 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
@@ -33,7 +27,6 @@ import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
@@ -43,8 +36,6 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
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.util.StringUtils;
/**
@@ -71,11 +62,10 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Lei Chen
* @author Eduard Domínguez
* @author Chris Bono
*/
public class KeyValueSerdeResolver implements ApplicationContextAware {
private static final Log LOG = LogFactory.getLog(KeyValueSerdeResolver.class);
private final Map<String, Object> streamConfigGlobalProperties;
private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
@@ -262,7 +252,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
Serde<?> fallbackSerde = getFallbackSerde("default.key.serde");
keySerde = getSerde(generic, fallbackSerde);
keySerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
}
if (keySerde == null) {
keySerde = Serdes.ByteArray();
@@ -286,101 +276,6 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
GlobalKTable.class.isAssignableFrom(resolvableType.getRawClass()));
}
private Serde<?> getSerde(ResolvableType generic, Serde<?> fallbackSerde) {
Serde<?> serde = null;
Map<String, Serde> beansOfType = context.getBeansOfType(Serde.class);
Serde<?>[] serdeBeans = new Serde<?>[1];
final Class<?> genericRawClazz = generic.getRawClass();
beansOfType.forEach((k, v) -> {
final Class<?> classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
context.getBeanFactory().getBeanDefinition(k))
.getMetadata().getClassName(),
ClassUtils.getDefaultClassLoader());
try {
Method[] methods = classObj.getMethods();
Optional<Method> serdeBeanMethod = Arrays.stream(methods).filter(m -> m.getName().equals(k)).findFirst();
if (serdeBeanMethod.isPresent()) {
Method method = serdeBeanMethod.get();
ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj);
ResolvableType serdeBeanGeneric = resolvableType.getGeneric(0);
Class<?> serdeGenericRawClazz = serdeBeanGeneric.getRawClass();
if (serdeGenericRawClazz != null && genericRawClazz != null) {
if (serdeGenericRawClazz.isAssignableFrom(genericRawClazz)) {
serdeBeans[0] = v;
}
}
}
}
catch (Exception e) {
// Pass through...
}
});
if (serdeBeans[0] != null) {
return serdeBeans[0];
}
if (genericRawClazz != null) {
if (Integer.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Integer();
}
else if (Long.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Long();
}
else if (Short.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Short();
}
else if (Double.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Double();
}
else if (Float.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Float();
}
else if (byte[].class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.ByteArray();
}
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);
}
}
}
return serde;
}
private 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;
}
}
return false;
}
private Serde<?> getValueSerde(String valueSerdeString, Map<String, ?> extendedConfiguration)
throws ClassNotFoundException {
Serde<?> valueSerde;
@@ -416,7 +311,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
(isResolvableKStreamArrayType(resolvableType)))) {
Serde<?> fallbackSerde = getFallbackSerde("default.value.serde");
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
valueSerde = getSerde(generic, fallbackSerde);
valueSerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
}
if (valueSerde == null) {

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

@@ -0,0 +1,241 @@
/*
* Copyright 2022-2022 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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.stream.Collectors;
import org.apache.commons.logging.Log;
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.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.util.ClassUtils;
/**
* Utility class that contains various methods to help resolve {@link Serde Serdes}.
*
* @author Chris Bono
* @since 4.0
*/
abstract class SerdeResolverUtils {
private static final Log LOG = LogFactory.getLog(SerdeResolverUtils.class);
/**
* 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.
*
* @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
*/
static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, 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();
}
else if (Long.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Long();
}
else if (Short.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Short();
}
else if (Double.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Double();
}
else if (Float.class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.Float();
}
else if (byte[].class.isAssignableFrom(genericRawClazz)) {
serde = Serdes.ByteArray();
}
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);
}
}
return serde;
}
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;
}
}
return false;
}
/**
* Find all {@link Serde} beans that are assignable from {@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
*/
static List<Serde<?>> findMatchingSerdes(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());
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);
}
}
});
if (!matchingSerdes.isEmpty()) {
return matchingSerdes.stream().sorted(Collections.reverseOrder())
.map(SerdeWithSpecificityScore::getSerde)
.collect(Collectors.toList());
}
return Collections.emptyList();
}
/**
* Calculate a score to indicate how specific of a match one resolvable type is to another.
* <p><br>Simple string comparison (the number of matching leading characters between two type strings) is used to
* calculate the score. This approach avoids the recursive nature of the possible generic types, and leverages
* the type strings returned from {@link ResolvableType#toString()} and {@link Type#getTypeName()} which already
* include the properly handled generic types. The score is a composite of both of these properties because the
* 'toString' value does not include bounds values and works as a tie-breaker to distinguish non-exact
* matches that are closer in nature.
* <p><br><b>Example:</b>
* <pre>{@code
* -------------------------------------------------------------------------------------------------------
* 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>'
* -------------------------------------------------------------------------------------------------------
* }</pre>
*
* If using only the 'toString' value then both types would have the same score. However, including the 'typeName'
* value in the score differentiates them - in this case it is clear that 'typeToCheck1' is a direct match.
*
* @param targetType the target type
* @param typeToScore the type to calculate a score for
* @return a score on how close of a match {@code typeToScore} is to {@code targetType} - the higher the score,
* the closer the match
*/
private static int calculateScore(ResolvableType targetType, ResolvableType typeToScore) {
int score = countLeadingMatchingChars(targetType.getType().getTypeName(), typeToScore.getType().getTypeName());
score += countLeadingMatchingChars(targetType.toString(), typeToScore.toString());
return score;
}
private static int countLeadingMatchingChars(String s1, String s2) {
if (s1 == null || s2 == null) {
return 0;
}
int matchCount = 0;
for (int i = 0; i < s1.length() && i < s2.length(); i++) {
if (s1.charAt(i) != s2.charAt(i)) {
break;
}
matchCount++;
}
return matchCount;
}
/**
* Private internal class used strictly to 'remember' a score for a serde and use it for sorting later.
*/
private static class SerdeWithSpecificityScore implements Comparable<SerdeWithSpecificityScore> {
private Integer score;
private Serde<?> serde;
SerdeWithSpecificityScore(Integer score, Serde<?> serde) {
this.score = Objects.requireNonNull(score);
this.serde = Objects.requireNonNull(serde);
}
Serde<?> getSerde() {
return serde;
}
@Override
public int compareTo(SerdeWithSpecificityScore other) {
return this.score.compareTo(other.score);
}
}
}

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

@@ -0,0 +1,149 @@
/*
* Copyright 2022-2022 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Date;
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.junit.jupiter.api.Test;
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 static org.assertj.core.api.Assertions.assertThat;
/**
* Unit tests for {@link SerdeResolverUtils}.
*
* @author Chris Bono
*/
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() {
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);
ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(SerdeResolverTestApp.class);
contextRunner.run((context) -> {
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geDate))
.extracting("name")
.containsExactly("genericEventDateSerde", "genericEventDateBoundedSerde", "genericEventWildcardSerde", "genericEventRawSerde");
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geBounded))
.extracting("name")
.containsExactly("genericEventDateBoundedSerde", "genericEventWildcardSerde");
assertThat(SerdeResolverUtils.findMatchingSerdes(context, geWildcard))
.extracting("name")
.containsExactly("genericEventWildcardSerde");
// 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 class GenericEventSerde<T> implements Serde<GenericEvent<? extends T>> {
private String name;
GenericEventSerde(String name) {
this.name = name;
}
String getName() {
return name;
}
@Override
public Serializer<GenericEvent<? extends T>> serializer() {
return null;
}
@Override
public Deserializer<GenericEvent<? extends T>> deserializer() {
return null;
}
}
static class GenericEvent<T> { }
@EnableAutoConfiguration
static class SerdeResolverTestApp {
@Bean
public Serde<GenericEvent<Date>> genericEventDateSerde() {
return new GenericEventSerde("genericEventDateSerde");
}
@Bean
public Serde<GenericEvent<? extends Date>> genericEventDateBoundedSerde() {
return new GenericEventSerde("genericEventDateBoundedSerde");
}
@Bean
public Serde<GenericEvent<String>> genericEventStringSerde() {
return new GenericEventSerde("genericEventStringSerde");
}
@Bean
public Serde<GenericEvent<?>> genericEventWildcardSerde() {
return new GenericEventSerde("genericEventWildcardSerde");
}
@Bean
public Serde<GenericEvent> genericEventRawSerde() {
return new GenericEventSerde("genericEventRawSerde");
}
@Bean
public Serde<?> widlcardSerde() {
return Serdes.Void();
}
}
}

169
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2019 the original author or authors.
* Copyright 2019-2022 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.
@@ -23,6 +23,7 @@ import java.util.function.Function;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.junit.ClassRule;
import org.junit.Test;
@@ -42,8 +43,15 @@ import org.springframework.context.annotation.Bean;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.util.Assert;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests to verify proper resolution of {@link Serde serdes} when they are provided as beans.
*
* @author Soby Chako
* @author Chris Bono
*/
public class SerdesProvidedAsBeansTests {
@ClassRule
@@ -52,44 +60,78 @@ public class SerdesProvidedAsBeansTests {
private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
@Test
public void testKstreamWordCountFunction() throws NoSuchMethodException {
SpringApplication app = new SpringApplication(SerdeProvidedAsBeanApp.class);
public void testSimpleSerdeBeansAreResolvedProperly() throws Exception {
SpringApplication app = new SpringApplication(SerdesProvidedAsBeansTestApp.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.process-in-0.destination=purchases",
"--spring.cloud.stream.bindings.process-out-0.destination=coffee",
"--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-0",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.function.definition=simpleProcess",
"--spring.cloud.stream.bindings.simpleProcess-in-0.destination=topic1",
"--spring.cloud.stream.bindings.simpleProcess-out-0.destination=topic2",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
final Method method = SerdeProvidedAsBeanApp.class.getMethod("process");
Method method = SerdesProvidedAsBeansTestApp.class.getMethod("simpleProcess");
ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdesProvidedAsBeansTestApp.class);
ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdeProvidedAsBeanApp.class);
KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
final KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
final BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
final KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("simpleProcess-in-0").getConsumer();
KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
final ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("process-in-0").getConsumer();
final KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
final Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
assertThat(inboundValueSerde).isInstanceOf(FooSerde.class);
Assert.isTrue(inboundValueSerde instanceof FooSerde, "Inbound Value Serde is not matched");
ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("simpleProcess-out-0").getProducer();
KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
final ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("process-out-0").getProducer();
final KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
final Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
assertThat(outboundValueSerde).isInstanceOf(FooSerde.class);
}
}
Assert.isTrue(outboundValueSerde instanceof FooSerde, "Outbound Value Serde is not matched");
@Test
public void testGenericSerdeBeansAreResolvedProperly() throws Exception {
SpringApplication app = new SpringApplication(SerdesProvidedAsBeansTestApp.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.function.definition=genericProcess",
"--spring.cloud.stream.bindings.genericProcess-in-0.destination=topic1",
"--spring.cloud.stream.bindings.genericProcess-out-0.destination=topic2",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
Method method = SerdesProvidedAsBeansTestApp.class.getMethod("genericProcess");
ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdesProvidedAsBeansTestApp.class);
KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("genericProcess-in-0").getConsumer();
KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
assertThat(inboundValueSerde).isInstanceOf(GenericEventDateSerde.class);
ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("genericProcess-out-0").getProducer();
KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
assertThat(outboundValueSerde).isInstanceOf(GenericEventStringSerde.class);
}
}
@@ -105,17 +147,76 @@ public class SerdesProvidedAsBeansTests {
}
}
static class GenericEventDateSerde implements Serde<GenericEvent<Date>> {
@Override
public Serializer<GenericEvent<Date>> serializer() {
return null;
}
@Override
public Deserializer<GenericEvent<Date>> deserializer() {
return null;
}
}
static class GenericEventStringSerde implements Serde<GenericEvent<String>> {
@Override
public Serializer<GenericEvent<String>> serializer() {
return null;
}
@Override
public Deserializer<GenericEvent<String>> deserializer() {
return null;
}
}
static class GenericEvent<T> {
static <X> GenericEvent<X> of(X newThing) {
GenericEvent<X> newEvent = new GenericEvent<>();
newEvent.setThing(newThing);
return newEvent;
}
private T thing;
public T getThing() {
return thing;
}
public void setThing(T thing) {
this.thing = thing;
}
}
@EnableAutoConfiguration
public static class SerdeProvidedAsBeanApp {
public static class SerdesProvidedAsBeansTestApp {
@Bean
public Function<KStream<String, Date>, KStream<String, Date>> process() {
return input -> input;
public Serde<GenericEvent<Date>> genericEventDataSerde() {
return new GenericEventDateSerde();
}
@Bean
public Serde<GenericEvent<String>> genericEventStringSerde() {
return new GenericEventStringSerde();
}
@Bean
public Serde<Date> fooSerde() {
return new FooSerde<>();
}
@Bean
public Function<KStream<String, Date>, KStream<String, Date>> simpleProcess() {
return input -> input;
}
@Bean
public Function<KStream<String, GenericEvent<Date>>, KStream<String, GenericEvent<String>>> genericProcess() {
return input -> input.map((k, v) -> new KeyValue(k, GenericEvent.of(v.toString())));
}
}
}