From ce6a03ee597acb58ea1f8630ba62374b2f09d57f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 6 Jun 2022 18:31:47 -0400 Subject: [PATCH] Kafk Streams binder message conversion issues When native decoding is disabled and message conversion is used in Kafka Streams binder, it doesn't currently carry the original headers forward. Fixing this issue. Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2411 --- .../AbstractKafkaStreamsBinderProcessor.java | 28 ++- ...fkaStreamsNativeEncodingDecodingTests.java | 191 ++++++++++++++++++ 2 files changed, 217 insertions(+), 2 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java 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)); + } + + } + +}