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

Kafka stream outbound content type polishing

Browse Source 
Using Jackson to write content type as bytes so that the outgoing
content type string is properly json compatible.
This commit is contained in:
Soby Chacko
2019-04-24 20:02:49 -04:00
committed by Oleg Zhurakousky
parent 1a19eeabec
commit 816a4ec232
1 changed files with 12 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -16,10 +16,10 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.header.Header;
@@ -113,9 +113,17 @@ public class KafkaStreamsMessageConversionDelegate {
public void process(Object key, Object value) {
if (perRecordContentTypeHolder.contentType != null) {
this.context.headers().remove(MessageHeaders.CONTENT_TYPE);
final Header header = new RecordHeader(MessageHeaders.CONTENT_TYPE, perRecordContentTypeHolder
.contentType.getBytes(StandardCharsets.UTF_8));
this.context.headers().add(header);
final Header header;
try {
header = new RecordHeader(MessageHeaders.CONTENT_TYPE,
new ObjectMapper().writeValueAsBytes(perRecordContentTypeHolder.contentType));
this.context.headers().add(header);
}
catch (Exception e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Could not add content type header");
}
}
perRecordContentTypeHolder.unsetContentType();
}
}