From 0b9e211e279458537e7226d488d4659521e59ed2 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 29 Dec 2017 16:16:41 -0500 Subject: [PATCH] Kafka Streams binder cleanup Aligning semantics of native encoding with core spring cloud stream. With this change, if nativeEncoding is set, any message conversion is skipped by the framework. It is up to the application to ensure that the data sent on the outbound is of type byte[], otherwise, it fails. If the nativeEncoding is false, which is the default, the binder does the message conversion before sending the data on the outbound. Corresponding changes in tests Resolves #267 --- .../stream/binder/kstream/KStreamBinder.java | 43 ++++++------------- .../kstream/KStreamBoundElementFactory.java | 19 +++++--- .../config/KStreamBinderConfiguration.java | 8 ++-- ...rPojoInputAndPrimitiveTypeOutputTests.java | 8 +--- ...StreamBinderWordCountIntegrationTests.java | 8 +--- ...treamInteractiveQueryIntegrationTests.java | 9 ++-- ...PojoInputStringOutputIntegrationTests.java | 5 +-- 7 files changed, 37 insertions(+), 63 deletions(-) diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java index a7ece547e..41bac7373 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java @@ -23,6 +23,7 @@ import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Produced; import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.Binding; @@ -30,7 +31,6 @@ import org.springframework.cloud.stream.binder.DefaultBinding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; @@ -42,6 +42,7 @@ import org.springframework.util.StringUtils; /** * @author Marius Bogoevici + * @author Soby Chacko */ public class KStreamBinder extends AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> @@ -53,11 +54,8 @@ public class KStreamBinder extends private final StreamsConfig streamsConfig; - private final KafkaBinderConfigurationProperties binderConfigurationProperties; - - public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, + public KStreamBinder(KafkaTopicProvisioner kafkaTopicProvisioner, KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { - this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; this.streamsConfig = streamsConfig; @@ -80,49 +78,32 @@ public class KStreamBinder extends ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties( new KafkaProducerProperties()); this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties); - if (!properties.isUseNativeEncoding()) { - outboundBindTarget = outboundBindTarget - .map((k, v) -> KeyValue.pair(k, (Object) KStreamBinder.this.serializePayloadIfNecessary((Message) v))); + outboundBindTarget = outboundBindTarget + .map((k, v) -> KeyValue.pair(k, ((Message) v).getPayload())); + + Serde keySerde = Serdes.ByteArray(); + Serde valueSerde = Serdes.ByteArray(); + if (properties.isUseNativeEncoding()) { + outboundBindTarget.to(name, Produced.with((Serde) keySerde, (Serde) valueSerde)); } else { - outboundBindTarget = outboundBindTarget - .map((k, v) -> KeyValue.pair(k, ((Message) v).getPayload())); - } - if (!properties.isUseNativeEncoding() || StringUtils.hasText(properties.getExtension().getKeySerde()) || StringUtils.hasText(properties.getExtension().getValueSerde())) { 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()); } } - 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) keySerde, (Serde) valueSerde, name); - } - catch (ClassNotFoundException e) { + outboundBindTarget.to(name, Produced.with((Serde) keySerde, (Serde) valueSerde)); + } catch (ClassNotFoundException e) { throw new IllegalStateException("Serde class not found: ", e); } - - } - else { - outboundBindTarget.to(name); } return new DefaultBinding<>(name, null, outboundBindTarget, null); } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java index 5296a8f92..2b53156c2 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.binder.kstream; +import java.util.HashMap; +import java.util.Map; + import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.kafka.streams.KeyValue; @@ -24,11 +27,11 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KeyValueMapper; import org.springframework.aop.framework.ProxyFactory; +import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; @@ -39,6 +42,7 @@ import org.springframework.util.StringUtils; /** * @author Marius Bogoevici + * @author Soby Chacko */ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory { @@ -103,8 +107,8 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory delegate) { Assert.notNull(delegate, "delegate cannot be null"); Assert.isNull(this.delegate, "delegate already set to " + this.delegate); - if (messageConverter != null) { + ProducerProperties producer = bindingServiceProperties.getBindingProperties(name).getProducer(); + + if (messageConverter != null && !producer.isUseNativeEncoding()) { KeyValueMapper> keyValueMapper = (k, v) -> { Message message = (Message) v; BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); String contentType = bindingProperties.getContentType(); - MutableMessageHeaders messageHeaders = new MutableMessageHeaders(((Message) v).getHeaders()); + Map headers = new HashMap<>(((Message) v).getHeaders()); if (!StringUtils.isEmpty(contentType)) { - messageHeaders.put(MessageHeaders.CONTENT_TYPE, contentType); + headers.put(MessageHeaders.CONTENT_TYPE, contentType); } + MessageHeaders messageHeaders = new MessageHeaders(headers); return new KeyValue<>(k, messageConverter.toMessage(message.getPayload(), messageHeaders)); diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java index 0c187bc2b..ded4b5169 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java @@ -32,6 +32,7 @@ import org.springframework.context.annotation.Configuration; /** * @author Marius Bogoevici * @author Gary Russell + * @author Soby Chacko */ @Configuration @EnableConfigurationProperties(KStreamExtendedBindingProperties.class) @@ -48,10 +49,9 @@ public class KStreamBinderConfiguration { } @Bean - public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, - KafkaTopicProvisioner kafkaTopicProvisioner, - KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { - return new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties, + public KStreamBinder kStreamBinder(KafkaTopicProvisioner kafkaTopicProvisioner, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { + return new KStreamBinder(kafkaTopicProvisioner, kStreamExtendedBindingProperties, streamsConfig); } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java index f3e785eac..ad96cae26 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java @@ -47,7 +47,6 @@ import org.springframework.messaging.handler.annotation.SendTo; import static org.assertj.core.api.Assertions.assertThat; /** - * * @author Soby Chacko * @author Gary Russell */ @@ -84,21 +83,18 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests { "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--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.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$ByteArraySerde", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { receiveAndValidateFoo(context); - } - finally { + } finally { context.close(); } } - private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{ + private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java index 089586a70..9d75e7f10 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java @@ -51,7 +51,6 @@ import org.springframework.messaging.handler.annotation.SendTo; import static org.assertj.core.api.Assertions.assertThat; /** - * * @author Marius Bogoevici * @author Soby Chacko * @author Gary Russell @@ -90,8 +89,6 @@ public class KStreamBinderWordCountIntegrationTests { "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--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.valueSerde=org.apache.kafka.common.serialization.Serdes$ByteArraySerde", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.timeWindow.length=5000", "--spring.cloud.stream.kstream.timeWindow.advanceBy=0", @@ -99,13 +96,12 @@ public class KStreamBinderWordCountIntegrationTests { "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { receiveAndValidate(context); - } - finally { + } finally { context.close(); } } - private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception{ + private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java index 38087dbae..ff76fe95e 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java @@ -86,21 +86,18 @@ public class KStreamInteractiveQueryIntegrationTests { "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$ByteArraySerde", "--spring.cloud.stream.bindings.output.producer.headerMode=raw", - "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { receiveAndValidateFoo(context); - } - finally { + } finally { context.close(); } } - private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{ + private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); @@ -149,7 +146,7 @@ public class KStreamInteractiveQueryIntegrationTests { KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams(); ReadOnlyKeyValueStore keyValueStore = streams.store("prod-id-count-store", QueryableStoreTypes.keyValueStore()); - return (Long)keyValueStore.get(id); + return (Long) keyValueStore.get(id); } } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java index f52fd2ec6..626747600 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java @@ -82,16 +82,13 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests { "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--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.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$ByteArraySerde", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { receiveAndValidateFoo(context); - } - finally { + } finally { context.close(); } }