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

Fixing concurrency issue in Kafka Streams binder

Browse Source 
For more details: e7c1c7f769
This commit is contained in:
Soby Chacko
2024-02-15 16:25:12 -05:00
parent 9e7d9f6c71
commit b7de020e16
1 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -90,7 +90,7 @@ public class KafkaStreamsMessageConversionDelegate {
String contentType = this.kstreamBindingInformationCatalogue
.getContentType(outboundBindTarget);
MessageConverter messageConverter = this.compositeMessageConverter;
final PerRecordContentTypeHolder perRecordContentTypeHolder = new PerRecordContentTypeHolder();
final ThreadLocal<PerRecordContentTypeHolder> perRecordContentTypeHolderThreadLocal = ThreadLocal.withInitial(PerRecordContentTypeHolder::new);
final KStream<?, ?> kStreamWithEnrichedHeaders = outboundBindTarget
.filter((k, v) -> v != null)
@@ -103,7 +103,7 @@ public class KafkaStreamsMessageConversionDelegate {
}
MessageHeaders messageHeaders = new MessageHeaders(headers);
final Message<?> convertedMessage = messageConverter.toMessage(message.getPayload(), messageHeaders);
perRecordContentTypeHolder.setContentType((String) messageHeaders.get(MessageHeaders.CONTENT_TYPE));
perRecordContentTypeHolderThreadLocal.get().setContentType((String) messageHeaders.get(MessageHeaders.CONTENT_TYPE));
return convertedMessage.getPayload();
});
@@ -118,12 +118,12 @@ public class KafkaStreamsMessageConversionDelegate {
@Override
public void process(Object key, Object value) {
if (perRecordContentTypeHolder.contentType != null) {
if (perRecordContentTypeHolderThreadLocal.get().contentType != null) {
this.context.headers().remove(MessageHeaders.CONTENT_TYPE);
final Header header;
try {
header = new RecordHeader(MessageHeaders.CONTENT_TYPE,
new ObjectMapper().writeValueAsBytes(perRecordContentTypeHolder.contentType));
new ObjectMapper().writeValueAsBytes(perRecordContentTypeHolderThreadLocal.get().contentType));
this.context.headers().add(header);
}
catch (Exception e) {
@@ -131,7 +131,7 @@ public class KafkaStreamsMessageConversionDelegate {
LOG.debug("Could not add content type header");
}
}
perRecordContentTypeHolder.unsetContentType();
perRecordContentTypeHolderThreadLocal.get().unsetContentType();
}
}