SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka Streams binder message conversion issues

Browse Source 
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
This commit is contained in:
Soby Chacko
2022-06-06 18:31:47 -04:00
parent eb6ea6c18f
commit 23087e0597
2 changed files with 41 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -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<Headers> headersAtomicReference = new AtomicReference<>();
stream.process((ProcessorSupplier<Object, Object, Void, Void>) () -> new Processor<Object, Object, Void, Void>() {
@Override
public void init(ProcessorContext<Void, Void> context) {
Processor.super.init(context);
}
@Override
public void process(Record<Object, Object> 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<String, Object> 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 {

17
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java
View File

@@ -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<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("decode-words");
template.sendDefault("foobar");
Message<String> 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<Header> 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));