diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java
index 59374cdd5..0fcc077c0 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java
@@ -24,6 +24,9 @@ import org.apache.kafka.streams.processor.api.ProcessorContext;
 import org.apache.kafka.streams.processor.api.Record;
 
 import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -117,7 +120,9 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
 		return (r, e) -> {
 			StreamBridge streamBridge = this.dltPublishingContext.getStreamBridge();
 			if (streamBridge != null) {
-				streamBridge.send(this.dltDestination, r.value());
+				Message<VIn> message = MessageBuilder.withPayload(r.value())
+					.setHeader(KafkaHeaders.KEY, r.key()).build();
+				streamBridge.send(this.dltDestination, message);
 			}
 		};
 	}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java
index d88944c58..5ce3e28e6 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java
@@ -21,6 +21,9 @@ 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.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.streams.kstream.KStream;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
@@ -57,6 +60,7 @@ public class DltAwareProcessorTests {
 	public static void setUp() {
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
 			embeddedKafka);
+		consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
 			consumerProps);
@@ -77,6 +81,7 @@ public class DltAwareProcessorTests {
 
 		try (ConfigurableApplicationContext context = app.run("--server.port=0",
 			"--spring.cloud.stream.bindings.errorStream-in-0.destination=error-stream-in",
+			"--spring.cloud.stream.kafka.bindings.hello-dlt-1.producer.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer",
 			"--spring.cloud.stream.kafka.streams.bindings.errorStream-in-0.consumer.application-id=test-error-stream-app-id",
 			"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
 			"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde"
@@ -91,15 +96,17 @@ public class DltAwareProcessorTests {
 
 	private void receiveAndValidate(String in, String out) {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
-		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
+		senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(
 			senderProps);
 		try {
-			KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+			KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
 			template.setDefaultTopic(in);
-			template.sendDefault("foobar");
+			template.send(in, "my-key", "foobar");
 			ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
 				out);
 			assertThat(cr.value().contains("foobar")).isTrue();
+			assertThat(cr.key().contains("my-key")).isTrue();
 		}
 		finally {
 			pf.destroy();
diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc
index a983b25f1..d425e6f2f 100644
--- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc
+++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc
@@ -189,3 +189,17 @@ public java.util.function.Function<KStream<String, String>, KStream<String, Stri
 ```
 
 In this case, when the record fails, the `DltAwareProcessor`, instead of using its built-in recoverer which publishes to a DLT, uses the user provided recoverer which is a `BiConsumer` that takes the failed record and the exception thrown as arguments.
+
+=== Handling Record Keys in DltAwareProcessor
+
+When sending failed records to a DLT using `DltAwareProcessor`, if you want to send the record keys to the DLT topic, then you need to set the proper serializer on the DLT binding.
+This is because, `DltAwareProcessor` uses `StreamBridge` which uses the regular Kafka binder (message-channel based) which by default uses a `ByteArraySerializer` for keys.
+In the case of record values, Spring Cloud Stream converts the payload to proper `byte[]`; however, that is not the case with keys, as it simply pass along what it received in the header as a key.
+If you are providing a non-byte array key, then that might cause class cast exceptions and to avoid that you need to set a serializer on the DLT binding as below.
+
+Assuming that the DLT destination is `hello-dlt-1` and the record key is of String datatype.
+
+```
+spring.cloud.stream.kafka.bindings.hello-dlt-1.producer.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+```
+