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(); } }