diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index a7680e960..b0158c921 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -44,6 +44,8 @@ import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.processor.TimestampExtractor; import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.ProcessorSupplier; import org.apache.kafka.streams.processor.api.Record; import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.state.KeyValueStore; @@ -468,11 +470,33 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application private KStream getkStream(BindingProperties bindingProperties, KStream stream, boolean nativeDecoding) { if (!nativeDecoding) { + AtomicReference headersAtomicReference = new AtomicReference<>(); + stream.process((ProcessorSupplier) () -> new Processor() { + + @Override + public void init(ProcessorContext context) { + Processor.super.init(context); + } + + @Override + public void process(Record record) { + final Headers headers = record.headers(); + headersAtomicReference.set(headers); + } + + @Override + public void close() { + Processor.super.close(); + } + }); stream = stream.mapValues((value) -> { Object returnValue; String contentType = bindingProperties.getContentType(); - if (value != null && !StringUtils.isEmpty(contentType)) { - returnValue = MessageBuilder.withPayload(value) + if (value != null && !StringUtils.hasText(contentType)) { + final Headers headers = headersAtomicReference.get(); + final Map headersMap = new HashMap<>(); + headers.forEach(header -> headersMap.put(header.key(), header.value())); + returnValue = MessageBuilder.withPayload(value).copyHeaders(headersMap) .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); } else { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java new file mode 100644 index 000000000..ddbc2d614 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java @@ -0,0 +1,191 @@ +/* + * Copyright 2018-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams.integration; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.Map; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Grouped; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +@RunWith(SpringRunner.class) +@ContextConfiguration +@DirtiesContext +public abstract class KafkaStreamsNativeEncodingDecodingTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "decode-counts", "decode-counts-1"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); + + @SpyBean + org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate; + + private static Consumer consumer; + + @BeforeClass + public static void setUp() { + System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers", + embeddedKafka.getBrokersAsString()); + System.setProperty("server.port", "0"); + System.setProperty("spring.jmx.enabled", "false"); + + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( + consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts", "decode-counts-1"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers"); + System.clearProperty("server.port"); + System.clearProperty("spring.jmx.enabled"); + } + + @SpringBootTest(classes = WordCountProcessorApplication.class, properties = { + "spring.cloud.stream.bindings.process-in-0.destination=decode-words-1", + "spring.cloud.stream.bindings.process-out-0.destination=decode-counts-1", + "spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId" + + "=NativeEncodingDecodingEnabledTests-abc" }, webEnvironment = SpringBootTest.WebEnvironment.NONE) + public static class NativeEncodingDecodingEnabledTests + extends KafkaStreamsNativeEncodingDecodingTests { + + @Test + public void test() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( + senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("decode-words-1"); + template.sendDefault("foobar"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, + "decode-counts-1"); + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + + verify(conversionDelegate, never()).serializeOnOutbound(any(KStream.class)); + verify(conversionDelegate, never()).deserializeOnInbound(any(Class.class), + any(KStream.class)); + } + + } + + @SpringBootTest(classes = WordCountProcessorApplication.class, webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "spring.cloud.stream.bindings.process-in-0.destination=decode-words", + "spring.cloud.stream.bindings.process-out-0.destination=decode-counts", + "spring.cloud.stream.bindings.process-in-0.consumer.useNativeDecoding=false", + "spring.cloud.stream.bindings.process-out-0.producer.useNativeEncoding=false", + "spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId" + + "=hello-NativeEncodingDecodingEnabledTests-xyz" }) + public static class NativeEncodingDecodingDisabledTests + extends KafkaStreamsNativeEncodingDecodingTests { + + @Test + public void test() { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( + senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("decode-words"); + Message msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build(); + template.send(msg); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, + "decode-counts"); + + final Headers headers = cr.headers(); + final Iterable
foo = headers.headers("foo"); + assertThat(foo.iterator().hasNext()).isTrue(); + final Header fooHeader = foo.iterator().next(); + assertThat(fooHeader.value()).isEqualTo("bar".getBytes(StandardCharsets.UTF_8)); + + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + + verify(conversionDelegate).serializeOnOutbound(any(KStream.class)); + verify(conversionDelegate).deserializeOnInbound(any(Class.class), + any(KStream.class)); + } + + } + + @EnableAutoConfiguration + public static class WordCountProcessorApplication { + + @Bean + public Function, KStream> process() { + + return input -> input + .flatMapValues( + value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .map((key, value) -> new KeyValue<>(value, value)) + .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) + .windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x")) + .toStream().map((key, value) -> new KeyValue<>(null, + "Count for " + key.key() + " : " + value)); + } + + } + +}