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

Infer SerDes used on input/output when possible

Browse Source 
When using native de/serializaion (which is now the default), the binder should infer the common Serde's
used on input and output. The Serdes inferred are - Integer, Long, Short, Double, Float, String, byte[] and
Spring Kafka provided JsonSerde.

Resolves #368

Address PR review comments

Addressing PR review comments

Resolves #672
This commit is contained in:
Soby Chacko
2019-06-13 12:02:20 -04:00
committed by Oleg Zhurakousky
parent c1db3d950c
commit 5dac51df62
21 changed files with 349 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-2019 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.
@@ -21,6 +21,7 @@ import java.util.Map;
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.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
@@ -120,9 +121,15 @@ class KStreamBinder extends
this.kafkaTopicProvisioner.provisionProducerDestination(name,
extendedProducerProperties);
Serde<?> keySerde = this.keyValueSerdeResolver
.getOuboundKeySerde(properties.getExtension());
Serde<?> valueSerde = this.keyValueSerdeResolver.getOutboundValueSerde(properties,
properties.getExtension());
.getOuboundKeySerde(properties.getExtension(), kafkaStreamsBindingInformationCatalogue.getOutboundKStreamResolvable());
Serde<?> valueSerde;
if (properties.isUseNativeEncoding()) {
valueSerde = this.keyValueSerdeResolver.getOutboundValueSerde(properties,
properties.getExtension(), kafkaStreamsBindingInformationCatalogue.getOutboundKStreamResolvable());
}
else {
valueSerde = Serdes.ByteArray();
}
to(properties.isUseNativeEncoding(), name, outboundBindTarget,
(Serde<Object>) keySerde, (Serde<Object>) valueSerde);
return new DefaultBinding<>(name, null, outboundBindTarget, null);

10
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java
View File

@@ -27,6 +27,7 @@ import org.apache.kafka.streams.kstream.KStream;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
/**
@@ -46,6 +47,8 @@ class KafkaStreamsBindingInformationCatalogue {
private final Set<StreamsBuilderFactoryBean> streamsBuilderFactoryBeans = new HashSet<>();
private ResolvableType outboundKStreamResolvable;
/**
* For a given bounded {@link KStream}, retrieve it's corresponding destination on the
* broker.
@@ -122,4 +125,11 @@ class KafkaStreamsBindingInformationCatalogue {
return this.streamsBuilderFactoryBeans;
}
public void setOutboundKStreamResolvable(ResolvableType outboundResolvable) {
this.outboundKStreamResolvable = outboundResolvable;
}
public ResolvableType getOutboundKStreamResolvable() {
return outboundKStreamResolvable;
}
}

31
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java
View File

@@ -30,6 +30,7 @@ import java.util.function.Function;
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.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
@@ -94,6 +95,8 @@ public class KafkaStreamsFunctionProcessor implements ApplicationContextAware {
private Set<String> origInputs = new TreeSet<>();
private Set<String> origOutputs = new TreeSet<>();
private ResolvableType outboundResolvableType;
public KafkaStreamsFunctionProcessor(BindingServiceProperties bindingServiceProperties,
KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties,
KeyValueSerdeResolver keyValueSerdeResolver,
@@ -132,18 +135,20 @@ public class KafkaStreamsFunctionProcessor implements ApplicationContextAware {
resolvableTypeMap.put(next, resolvableType.getGeneric(0));
origInputs.remove(next);
ResolvableType iterableResType = resolvableType;
for (int i = 1; i < inputCount; i++) {
if (iterator.hasNext()) {
ResolvableType generic = resolvableType.getGeneric(1);
if (generic.getRawClass() != null &&
(generic.getRawClass().equals(Function.class) ||
generic.getRawClass().equals(Consumer.class))) {
iterableResType = iterableResType.getGeneric(1);
if (iterableResType.getRawClass() != null &&
(iterableResType.getRawClass().equals(Function.class) ||
iterableResType.getRawClass().equals(Consumer.class))) {
final String next1 = iterator.next();
resolvableTypeMap.put(next1, generic.getGeneric(0));
resolvableTypeMap.put(next1, iterableResType.getGeneric(0));
origInputs.remove(next1);
}
}
}
outboundResolvableType = iterableResType.getGeneric(1);
return resolvableTypeMap;
}
@@ -184,6 +189,8 @@ public class KafkaStreamsFunctionProcessor implements ApplicationContextAware {
i++;
}
if (result != null) {
kafkaStreamsBindingInformationCatalogue.setOutboundKStreamResolvable(
outboundResolvableType != null ? outboundResolvableType : resolvableType.getGeneric(1));
final Set<String> outputs = new TreeSet<>(origOutputs);
final Iterator<String> iterator = outputs.iterator();
@@ -251,9 +258,17 @@ public class KafkaStreamsFunctionProcessor implements ApplicationContextAware {
KafkaStreamsConsumerProperties extendedConsumerProperties =
this.kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(input);
//get state store spec
Serde<?> keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties);
Serde<?> valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(
bindingProperties.getConsumer(), extendedConsumerProperties);
Serde<?> keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties, stringResolvableTypeMap.get(input));
Serde<?> valueSerde;
if (bindingServiceProperties.getConsumerProperties(input).isUseNativeDecoding()) {
valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(
bindingProperties.getConsumer(), extendedConsumerProperties, stringResolvableTypeMap.get(input));
}
else {
valueSerde = Serdes.ByteArray();
}
final KafkaConsumerProperties.StartOffset startOffset = extendedConsumerProperties.getStartOffset();
Topology.AutoOffsetReset autoOffsetReset = null;

18
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
View File

@@ -28,6 +28,7 @@ import java.util.Properties;
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.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
@@ -64,6 +65,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
@@ -190,6 +192,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator
Assert.isTrue(methodAnnotatedOutboundNames.length == 1,
"Result does not match with the number of declared outbounds");
}
kafkaStreamsBindingInformationCatalogue.setOutboundKStreamResolvable(ResolvableType.forMethodReturnType(method));
if (result.getClass().isArray()) {
Object[] outboundKStreams = (Object[]) result;
int i = 0;
@@ -268,10 +271,19 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator
.getExtendedConsumerProperties(inboundName);
// get state store spec
KafkaStreamsStateStoreProperties spec = buildStateStoreSpec(method);
Serde<?> keySerde = this.keyValueSerdeResolver
.getInboundKeySerde(extendedConsumerProperties);
Serde<?> valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(
bindingProperties.getConsumer(), extendedConsumerProperties);
.getInboundKeySerde(extendedConsumerProperties, ResolvableType.forMethodParameter(methodParameter));
Serde<?> valueSerde;
if (bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding()) {
valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(
bindingProperties.getConsumer(), extendedConsumerProperties, ResolvableType.forMethodParameter(methodParameter));
}
else {
//keySerde = Serdes.ByteArray();
valueSerde = Serdes.ByteArray();
}
final KafkaConsumerProperties.StartOffset startOffset = extendedConsumerProperties
.getStartOffset();

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

@@ -21,12 +21,17 @@ import java.util.Map;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.util.StringUtils;
/**
@@ -78,6 +83,13 @@ public class KeyValueSerdeResolver {
return getKeySerde(keySerdeString);
}
public Serde<?> getInboundKeySerde(
KafkaStreamsConsumerProperties extendedConsumerProperties, ResolvableType resolvableType) {
String keySerdeString = extendedConsumerProperties.getKeySerde();
return getKeySerde(keySerdeString, resolvableType);
}
/**
* Provide the {@link Serde} for inbound value.
* @param consumerProperties {@link ConsumerProperties} on binding
@@ -105,6 +117,27 @@ public class KeyValueSerdeResolver {
return valueSerde;
}
public Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties,
KafkaStreamsConsumerProperties extendedConsumerProperties,
ResolvableType resolvableType) {
Serde<?> valueSerde;
String valueSerdeString = extendedConsumerProperties.getValueSerde();
try {
if (consumerProperties != null && consumerProperties.isUseNativeDecoding()) {
valueSerde = getValueSerde(valueSerdeString, resolvableType);
}
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(this.streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return valueSerde;
}
/**
* Provide the {@link Serde} for outbound key.
* @param properties binding level extended {@link KafkaStreamsProducerProperties}
@@ -114,6 +147,11 @@ public class KeyValueSerdeResolver {
return getKeySerde(properties.getKeySerde());
}
public Serde<?> getOuboundKeySerde(KafkaStreamsProducerProperties properties, ResolvableType resolvableType) {
return getKeySerde(properties.getKeySerde(), resolvableType);
}
/**
* Provide the {@link Serde} for outbound value.
* @param producerProperties {@link ProducerProperties} on binding
@@ -140,6 +178,25 @@ public class KeyValueSerdeResolver {
return valueSerde;
}
public Serde<?> getOutboundValueSerde(ProducerProperties producerProperties,
KafkaStreamsProducerProperties kafkaStreamsProducerProperties, ResolvableType resolvableType) {
Serde<?> valueSerde;
try {
if (producerProperties.isUseNativeEncoding()) {
valueSerde = getValueSerde(
kafkaStreamsProducerProperties.getValueSerde(), resolvableType);
}
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(this.streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return valueSerde;
}
/**
* Provide the {@link Serde} for state store.
* @param keySerdeString serde class used for key
@@ -186,6 +243,76 @@ public class KeyValueSerdeResolver {
return keySerde;
}
private Serde<?> getKeySerde(String keySerdeString, ResolvableType resolvableType) {
Serde<?> keySerde = null;
try {
if (StringUtils.hasText(keySerdeString)) {
keySerde = Utils.newInstance(keySerdeString, Serde.class);
}
else {
if (resolvableType != null &&
(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
keySerde = getSerde(keySerde, generic);
}
if (keySerde == null) {
keySerde = this.binderConfigurationProperties.getConfiguration()
.containsKey("default.key.serde")
? Utils.newInstance(this.binderConfigurationProperties
.getConfiguration().get("default.key.serde"),
Serde.class)
: Serdes.ByteArray();
}
}
keySerde.configure(this.streamConfigGlobalProperties, true);
}
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return keySerde;
}
private boolean isResolvableKStreamArrayType(ResolvableType resolvableType) {
return resolvableType.isArray() &&
KStream.class.isAssignableFrom(resolvableType.getComponentType().getRawClass());
}
private boolean isResolvalbeKafkaStreamsType(ResolvableType resolvableType) {
return resolvableType.getRawClass() != null && (KStream.class.isAssignableFrom(resolvableType.getRawClass()) || KTable.class.isAssignableFrom(resolvableType.getRawClass()) ||
GlobalKTable.class.isAssignableFrom(resolvableType.getRawClass()));
}
private Serde<?> getSerde(Serde<?> keySerde, ResolvableType generic) {
if (generic.getRawClass() != null) {
if (Integer.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.Integer();
}
else if (Long.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.Long();
}
else if (Short.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.Short();
}
else if (Double.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.Double();
}
else if (Float.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.Float();
}
else if (byte[].class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.ByteArray();
}
else if (String.class.isAssignableFrom(generic.getRawClass())) {
keySerde = Serdes.String();
}
else {
keySerde = new JsonSerde(generic.getRawClass());
}
}
return keySerde;
}
private Serde<?> getValueSerde(String valueSerdeString)
throws ClassNotFoundException {
Serde<?> valueSerde;
@@ -203,4 +330,32 @@ public class KeyValueSerdeResolver {
return valueSerde;
}
@SuppressWarnings("unchecked")
private Serde<?> getValueSerde(String valueSerdeString, ResolvableType resolvableType)
throws ClassNotFoundException {
Serde<?> valueSerde = null;
if (StringUtils.hasText(valueSerdeString)) {
valueSerde = Utils.newInstance(valueSerdeString, Serde.class);
}
else {
if (resolvableType != null && ((isResolvalbeKafkaStreamsType(resolvableType)) ||
(isResolvableKStreamArrayType(resolvableType)))) {
ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
valueSerde = getSerde(valueSerde, generic);
}
if (valueSerde == null) {
valueSerde = this.binderConfigurationProperties.getConfiguration()
.containsKey("default.value.serde")
? Utils.newInstance(this.binderConfigurationProperties
.getConfiguration().get("default.value.serde"),
Serde.class)
: Serdes.ByteArray();
}
}
return valueSerde;
}
}

6
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java
View File

@@ -94,9 +94,9 @@ public class KafkaStreamsBinderWordCountBranchesFunctionTests {
"=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.bindings.output1.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output1.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.timeWindow.length=5000",
"--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId" +

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -93,7 +93,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
"=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.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
//"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
receiveAndValidate(context);
}

45
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java
View File

@@ -49,7 +49,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -76,26 +76,6 @@ public class StreamToGlobalKTableFunctionTests {
"--spring.cloud.stream.bindings.input-x.destination=customers",
"--spring.cloud.stream.bindings.input-y.destination=products",
"--spring.cloud.stream.bindings.output.destination=enriched-order",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde" +
"=org.springframework.cloud.stream.binder.kafka.streams.function" +
".StreamToGlobalKTableFunctionTests$OrderSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.valueSerde" +
"=org.springframework.cloud.stream.binder.kafka.streams.function" +
".StreamToGlobalKTableFunctionTests$CustomerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.valueSerde" +
"=org.springframework.cloud.stream.binder.kafka.streams.function" +
".StreamToGlobalKTableFunctionTests$ProductSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde" +
"=org.springframework.cloud.stream.binder.kafka.streams." +
"function.StreamToGlobalKTableFunctionTests$EnrichedOrderSerde",
"--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" +
@@ -107,9 +87,8 @@ public class StreamToGlobalKTableFunctionTests {
"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) {
Map<String, Object> senderPropsCustomer = KafkaTestUtils.producerProps(embeddedKafka);
senderPropsCustomer.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
CustomerSerde customerSerde = new CustomerSerde();
senderPropsCustomer.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
customerSerde.serializer().getClass());
JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Customer> pfCustomer =
new DefaultKafkaProducerFactory<>(senderPropsCustomer);
@@ -123,8 +102,7 @@ public class StreamToGlobalKTableFunctionTests {
Map<String, Object> senderPropsProduct = KafkaTestUtils.producerProps(embeddedKafka);
senderPropsProduct.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
ProductSerde productSerde = new ProductSerde();
senderPropsProduct.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, productSerde.serializer().getClass());
senderPropsProduct.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Product> pfProduct =
new DefaultKafkaProducerFactory<>(senderPropsProduct);
@@ -139,8 +117,7 @@ public class StreamToGlobalKTableFunctionTests {
Map<String, Object> senderPropsOrder = KafkaTestUtils.producerProps(embeddedKafka);
senderPropsOrder.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
OrderSerde orderSerde = new OrderSerde();
senderPropsOrder.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, orderSerde.serializer().getClass());
senderPropsOrder.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Order> pfOrder = new DefaultKafkaProducerFactory<>(senderPropsOrder);
KafkaTemplate<Long, Order> orderTemplate = new KafkaTemplate<>(pfOrder, true);
@@ -157,9 +134,8 @@ public class StreamToGlobalKTableFunctionTests {
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
EnrichedOrderSerde enrichedOrderSerde = new EnrichedOrderSerde();
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
enrichedOrderSerde.deserializer().getClass());
JsonDeserializer.class);
consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE,
"org.springframework.cloud.stream.binder.kafka.streams." +
"function.StreamToGlobalKTableFunctionTests.EnrichedOrder");
@@ -334,15 +310,4 @@ public class StreamToGlobalKTableFunctionTests {
}
}
public static class OrderSerde extends JsonSerde<Order> {
}
public static class CustomerSerde extends JsonSerde<Customer> {
}
public static class ProductSerde extends JsonSerde<Product> {
}
public static class EnrichedOrderSerde extends JsonSerde<EnrichedOrder> {
}
}

24
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java
View File

@@ -88,18 +88,6 @@ public class StreamToTableJoinFunctionTests {
"--spring.cloud.stream.bindings.input-1.destination=user-clicks-1",
"--spring.cloud.stream.bindings.input-2.destination=user-regions-1",
"--spring.cloud.stream.bindings.output.destination=output-topic-1",
"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--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" +
@@ -236,18 +224,6 @@ public class StreamToTableJoinFunctionTests {
"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
"--spring.cloud.stream.kafka.streams.binder.configuration.auto.offset.reset=latest",
"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.startOffset=earliest",
"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde" +
"=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--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" +

7
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java
View File

@@ -109,7 +109,7 @@ public abstract class DeserializationErrorHandlerByKafkaTests {
"spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=deser-kafka-dlq",
"spring.cloud.stream.bindings.input.group=group",
"spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq",
"spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde="
"spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde" }, webEnvironment = SpringBootTest.WebEnvironment.NONE)
// @checkstyle:on
public static class DeserializationByKafkaAndDlqTests
@@ -148,10 +148,11 @@ public abstract class DeserializationErrorHandlerByKafkaTests {
// @checkstyle:off
@SpringBootTest(properties = {
"spring.cloud.stream.bindings.input.destination=word1,word2",
"spring.cloud.stream.kafka.streams.default.consumer.applicationId=deser-kafka-dlq-multi-input",
"spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=deser-kafka-dlq-multi-input",
//"spring.cloud.stream.kafka.streams.default.consumer.applicationId=deser-kafka-dlq-multi-input",
"spring.cloud.stream.bindings.input.group=groupx",
"spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq",
"spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde="
"spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde" }, webEnvironment = SpringBootTest.WebEnvironment.NONE)
// @checkstyle:on
public static class DeserializationByKafkaAndDlqTestsWithMultipleInputs

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java
View File

@@ -111,8 +111,8 @@ public abstract class DeserializtionErrorHandlerByBinderTests {
+ "=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.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq",
"spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id"
+ "=deserializationByBinderAndDlqTests",
@@ -160,8 +160,8 @@ public abstract class DeserializtionErrorHandlerByBinderTests {
+ "=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.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq",
"spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id"
+ "=deserializationByBinderAndDlqTestsWithMultipleInputs",

50
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java
View File

@@ -207,14 +207,14 @@ public class KafkaStreamsBinderHealthIndicatorTests {
+ "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.bindings.output.producer.keySerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde="
// + "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId="
+ "ApplicationHealthTest-xyz",
"--spring.cloud.stream.kafka.streams.binder.brokers="
@@ -237,23 +237,23 @@ public class KafkaStreamsBinderHealthIndicatorTests {
+ "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.bindings.output.producer.keySerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.bindings.output2.producer.keySerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input2.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output2.producer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input2.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde="
// + "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output2.producer.keySerde="
// + "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
//
// "--spring.cloud.stream.kafka.streams.bindings.input2.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output2.producer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input2.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderHealthIndicatorTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId="
+ "ApplicationHealthTest-xyz",

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java
View File

@@ -108,7 +108,7 @@ public class KafkaStreamsBinderMultipleInputTopicsTest {
+ "=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.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.timeWindow.length=5000",
"--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId"

14
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java
View File

@@ -97,13 +97,13 @@ public class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
+ "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.bindings.output.producer.keySerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde="
+ "org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde="
// + "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde="
// + "org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId="
+ "KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests-xyz",
"--spring.cloud.stream.kafka.streams.binder.brokers="

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java
View File

@@ -114,7 +114,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests {
+ "=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.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
//"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.timeWindow.length=5000",
"--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0",
"--spring.cloud.stream.kafka.streams.binder.brokers="

6
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsInteractiveQueryIntegrationTests.java
View File

@@ -100,9 +100,9 @@ public class KafkaStreamsInteractiveQueryIntegrationTests {
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsInteractiveQueryIntegrationTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsInteractiveQueryIntegrationTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-abc",
"--spring.cloud.stream.kafka.streams.binder.configuration.application.server"

18
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java
View File

@@ -67,9 +67,9 @@ public class KafkaStreamsStateStoreIntegrationTests {
+ "=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.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId"
+ "=KafkaStreamsStateStoreIntegrationTests-abc",
"--spring.cloud.stream.kafka.streams.binder.brokers="
@@ -100,12 +100,12 @@ public class KafkaStreamsStateStoreIntegrationTests {
+ "=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.bindings.input1.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input1.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input2.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input2.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input1.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input1.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.input2.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input2.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkaStreamsStateStoreIntegrationTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input1.consumer.applicationId"
+ "=KafkaStreamsStateStoreIntegrationTests-abc",
"--spring.cloud.stream.kafka.streams.binder.brokers="

10
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java
View File

@@ -97,11 +97,11 @@ public class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
+ "=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.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
"org.springframework.cloud.stream.binder.kafka.streams.integration.KafkastreamsBinderPojoInputStringOutputIntegrationTests.Product",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.configuration.spring.json.value.default.type=" +
// "org.springframework.cloud.stream.binder.kafka.streams.integration.KafkastreamsBinderPojoInputStringOutputIntegrationTests.Product",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-xyz",
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString(),

84
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java
View File

@@ -48,7 +48,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -81,26 +81,26 @@ public class StreamToGlobalKTableJoinIntegrationTests {
"--spring.cloud.stream.bindings.input-x.destination=customers",
"--spring.cloud.stream.bindings.input-y.destination=products",
"--spring.cloud.stream.bindings.output.destination=enriched-order",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
+ "=org.springframework.cloud.stream.binder.kafka.streams.integration"
+ ".StreamToGlobalKTableJoinIntegrationTests$OrderSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.valueSerde"
+ "=org.springframework.cloud.stream.binder.kafka.streams.integration."
+ "StreamToGlobalKTableJoinIntegrationTests$CustomerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.valueSerde"
+ "=org.springframework.cloud.stream.binder.kafka.streams."
+ "integration.StreamToGlobalKTableJoinIntegrationTests$ProductSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
+ "=org.springframework.cloud.stream.binder.kafka.streams.integration"
+ ".StreamToGlobalKTableJoinIntegrationTests$EnrichedOrderSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
// + "=org.springframework.cloud.stream.binder.kafka.streams.integration"
// + ".StreamToGlobalKTableJoinIntegrationTests$OrderSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.valueSerde"
// + "=org.springframework.cloud.stream.binder.kafka.streams.integration."
// + "StreamToGlobalKTableJoinIntegrationTests$CustomerSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.valueSerde"
// + "=org.springframework.cloud.stream.binder.kafka.streams."
// + "integration.StreamToGlobalKTableJoinIntegrationTests$ProductSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
// + "=org.springframework.cloud.stream.binder.kafka.streams.integration"
// + ".StreamToGlobalKTableJoinIntegrationTests$EnrichedOrderSerde",
"--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"
@@ -116,9 +116,8 @@ public class StreamToGlobalKTableJoinIntegrationTests {
.producerProps(embeddedKafka);
senderPropsCustomer.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
LongSerializer.class);
CustomerSerde customerSerde = new CustomerSerde();
senderPropsCustomer.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
customerSerde.serializer().getClass());
JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Customer> pfCustomer = new DefaultKafkaProducerFactory<>(
senderPropsCustomer);
@@ -135,9 +134,8 @@ public class StreamToGlobalKTableJoinIntegrationTests {
.producerProps(embeddedKafka);
senderPropsProduct.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
LongSerializer.class);
ProductSerde productSerde = new ProductSerde();
senderPropsProduct.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
productSerde.serializer().getClass());
JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Product> pfProduct = new DefaultKafkaProducerFactory<>(
senderPropsProduct);
@@ -155,9 +153,8 @@ public class StreamToGlobalKTableJoinIntegrationTests {
.producerProps(embeddedKafka);
senderPropsOrder.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
LongSerializer.class);
OrderSerde orderSerde = new OrderSerde();
senderPropsOrder.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
orderSerde.serializer().getClass());
JsonSerializer.class);
DefaultKafkaProducerFactory<Long, Order> pfOrder = new DefaultKafkaProducerFactory<>(
senderPropsOrder);
@@ -176,9 +173,8 @@ public class StreamToGlobalKTableJoinIntegrationTests {
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
LongDeserializer.class);
EnrichedOrderSerde enrichedOrderSerde = new EnrichedOrderSerde();
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
enrichedOrderSerde.deserializer().getClass());
JsonDeserializer.class);
consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE,
"org.springframework.cloud.stream.binder.kafka.streams.integration."
+ "StreamToGlobalKTableJoinIntegrationTests.EnrichedOrder");
@@ -368,20 +364,20 @@ public class StreamToGlobalKTableJoinIntegrationTests {
}
public static class OrderSerde extends JsonSerde<Order> {
}
public static class CustomerSerde extends JsonSerde<Customer> {
}
public static class ProductSerde extends JsonSerde<Product> {
}
public static class EnrichedOrderSerde extends JsonSerde<EnrichedOrder> {
}
// public static class OrderSerde extends JsonSerde<Order> {
//
// }
//
// public static class CustomerSerde extends JsonSerde<Customer> {
//
// }
//
// public static class ProductSerde extends JsonSerde<Product> {
//
// }
//
// public static class EnrichedOrderSerde extends JsonSerde<EnrichedOrder> {
//
// }
}

48
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java
View File

@@ -96,18 +96,18 @@ public class StreamToTableJoinIntegrationTests {
"--spring.cloud.stream.bindings.input-x.destination=user-regions-1",
"--spring.cloud.stream.bindings.output.destination=output-topic-1",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--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"
@@ -247,18 +247,18 @@ public class StreamToTableJoinIntegrationTests {
"--spring.cloud.stream.kafka.streams.binder.configuration.auto.offset.reset=latest",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.startOffset=earliest",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
+ "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
// "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde"
// + "=org.apache.kafka.common.serialization.Serdes$StringSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde"
// + "=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--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"

6
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/WordCountMultipleBranchesIntegrationTests.java
View File

@@ -103,9 +103,9 @@ public class WordCountMultipleBranchesIntegrationTests {
+ "=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.bindings.output1.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output1.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
// "--spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=org.springframework.kafka.support.serializer.JsonSerde",
"--spring.cloud.stream.kafka.streams.timeWindow.length=5000",
"--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId"