From 23087e0597d4100c9f608e156f3b068fe1229217 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 6 Jun 2022 18:31:47 -0400 Subject: [PATCH] Kafka 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 | 17 +++++++++-- 2 files changed, 41 insertions(+), 4 deletions(-) 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 index 4bb5abb46..f5f2d60bf 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * 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. @@ -16,6 +16,7 @@ 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; @@ -23,6 +24,8 @@ import java.util.Map; 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; @@ -47,7 +50,9 @@ 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.handler.annotation.SendTo; +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; @@ -146,7 +151,8 @@ public abstract class KafkaStreamsNativeEncodingDecodingTests { senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic("decode-words"); - template.sendDefault("foobar"); + Message msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build(); + template.send(msg); StopWatch stopWatch = new StopWatch(); stopWatch.start(); System.out.println("Starting: "); @@ -154,6 +160,13 @@ public abstract class KafkaStreamsNativeEncodingDecodingTests { "decode-counts"); stopWatch.stop(); System.out.println("Total time: " + stopWatch.getTotalTimeSeconds()); + + 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));