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

Kstream binder: producer default Serde changes

Browse Source 
Change the way the default Serde classes are selected for key and value
in producer when only one of those is provided by the user.

Fix #190
This commit is contained in:
Soby Chacko
2017-09-09 10:04:23 -04:00
committed by Gary Russell
parent 6485bd2abd
commit 1aecd02404
2 changed files with 28 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java
View File

@@ -62,8 +62,11 @@ public class KStreamBinder extends
private final StreamsConfig streamsConfig;
private final KafkaBinderConfigurationProperties binderConfigurationProperties;
public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner,
KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
this.binderConfigurationProperties = binderConfigurationProperties;
this.headers = EmbeddedHeaderUtils.headersToEmbed(binderConfigurationProperties.getHeaders());
this.kafkaTopicProvisioner = kafkaTopicProvisioner;
this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
@@ -72,7 +75,7 @@ public class KStreamBinder extends
@Override
protected Binding<KStream<Object, Object>> doBindConsumer(String name, String group,
KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<KafkaConsumerProperties>(
new KafkaConsumerProperties());
@@ -83,17 +86,17 @@ public class KStreamBinder extends
@Override
@SuppressWarnings("unchecked")
protected Binding<KStream<Object, Object>> doBindProducer(String name, KStream<Object, Object> outboundBindTarget,
ExtendedProducerProperties<KStreamProducerProperties> properties) {
ExtendedProducerProperties<KStreamProducerProperties> properties) {
ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<KafkaProducerProperties>(
new KafkaProducerProperties());
this.kafkaTopicProvisioner.provisionProducerDestination(name , extendedProducerProperties);
this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties);
if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
outboundBindTarget = outboundBindTarget.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
@Override
public KeyValue<Object, Object> apply(Object k, Object v) {
if (v instanceof Message) {
try {
return new KeyValue<>(k, (Object)KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
return new KeyValue<>(k, (Object) KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
}
catch (Exception e) {
throw new IllegalArgumentException(e);
@@ -111,7 +114,7 @@ public class KStreamBinder extends
.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
@Override
public KeyValue<Object, Object> apply(Object k, Object v) {
return KeyValue.pair(k, (Object)KStreamBinder.this.serializePayloadIfNecessary((Message<?>) v));
return KeyValue.pair(k, (Object) KStreamBinder.this.serializePayloadIfNecessary((Message<?>) v));
}
});
}
@@ -126,31 +129,37 @@ public class KStreamBinder extends
}
}
if (!properties.isUseNativeEncoding() || StringUtils.hasText(properties.getExtension().getKeySerde()) || StringUtils.hasText(properties.getExtension().getValueSerde())) {
Serde<?> keySerde = Serdes.ByteArray();
Serde<?> valueSerde = Serdes.ByteArray();
try {
Serde<?> keySerde;
Serde<?> valueSerde;
if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
if (keySerde instanceof Configurable) {
((Configurable) keySerde).configure(streamsConfig.originals());
}
}
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
try {
else {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
}
if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
if (valueSerde instanceof Configurable) {
((Configurable) valueSerde).configure(streamsConfig.originals());
}
}
else {
valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
}
outboundBindTarget.to((Serde<Object>) keySerde, (Serde<Object>) valueSerde, name);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
outboundBindTarget.to((Serde<Object>) keySerde, (Serde<Object>) valueSerde, name);
}
else {
outboundBindTarget.to(name);

9
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java
View File

@@ -85,6 +85,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
"--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.bindings.output.producer.headerMode=raw",
"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
"--spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.bindings.input.consumer.headerMode=raw",
"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
@@ -108,7 +109,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
@StreamListener("input")
@SendTo("output")
public KStream<?, String> process(KStream<Object, Product> input) {
public KStream<Integer, String> process(KStream<Object, Product> input) {
return input
.filter(new Predicate<Object, Product>() {
@@ -128,11 +129,11 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
.groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))
.count(TimeWindows.of(5000), "id-count-store")
.toStream()
.map(new KeyValueMapper<Windowed<Product>, Long, KeyValue<Object, String>>() {
.map(new KeyValueMapper<Windowed<Product>, Long, KeyValue<Integer, String>>() {
@Override
public KeyValue<Object, String> apply(Windowed<Product> key, Long value) {
return new KeyValue<>(null, "Count for product with ID 123: " + value);
public KeyValue<Integer, String> apply(Windowed<Product> key, Long value) {
return new KeyValue<>(key.key().id, "Count for product with ID 123: " + value);
}
});
}