diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
index 60cf9a5b8..4ffbdfcdb 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
@@ -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();
 				}
 			}