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<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 {
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<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));