diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index fe0d365a7..fde9d2d6e 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -1190,10 +1190,11 @@ public class KafkaMessageChannelBinder extends
 					.getDlqProducerProperties();
 			KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(
 					properties.getExtension().getTransactionManager());
+			final ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(dlqProducerProperties);
+			producerProperties.populateBindingName(properties.getBindingName());
 			ProducerFactory<?, ?> producerFactory = transMan != null
 					? transMan.getProducerFactory()
-					: getProducerFactory(null,
-							new ExtendedProducerProperties<>(dlqProducerProperties),
+					: getProducerFactory(null, producerProperties,
 							destination.getName() + ".dlq.producer", destination.getName());
 			final KafkaTemplate<?, ?> kafkaTemplate = new KafkaTemplate<>(
 					producerFactory);
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java
index 2002ee773..5dc0dbce0 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.kafka.integration;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 
@@ -28,6 +30,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.kafka.ListenerContainerWithDlqAndRetryCustomizer;
+import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer;
 import org.springframework.cloud.stream.binding.BindingsLifecycleController;
 import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.core.KafkaOperations;
@@ -48,6 +51,7 @@ import static org.mockito.Mockito.mock;
 
 /**
  * @author Gary Russell
+ * @author Soby Chacko
  */
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
 		"spring.cloud.function.definition=retryInBinder;retryInContainer",
@@ -59,8 +63,10 @@ import static org.mockito.Mockito.mock;
 @DirtiesContext
 public class KafkaRetryDlqBinderOrContainerTests {
 
+	private static final CountDownLatch latch = new CountDownLatch(2);
+
 	@Test
-	public void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) {
+	void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) throws Exception {
 		Binding<?> retryInBinder = controller.queryState("retryInBinder-in-0");
 		assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, "lifecycle.retryTemplate")).isNotNull();
 		assertThat(KafkaTestUtils.getPropertyValue(retryInBinder,
@@ -72,6 +78,9 @@ public class KafkaRetryDlqBinderOrContainerTests {
 		assertThat(KafkaTestUtils.getPropertyValue(retryInContainer,
 					"lifecycle.messageListenerContainer.commonErrorHandler.failureTracker.backOff"))
 				.isInstanceOf(ExponentialBackOffWithMaxRetries.class);
+		boolean await = latch.await(5, TimeUnit.SECONDS);
+		assertThat(await).isTrue();
+
 	}
 
 	@SpringBootApplication
@@ -112,6 +121,19 @@ public class KafkaRetryDlqBinderOrContainerTests {
 			};
 		}
 
+		// Because we have DLQ enabled on the consumer binding,
+		// this ProducerConfigCustomizer is used by the producer on DLQ.
+		@Bean
+		public ProducerConfigCustomizer producerConfigCustomizer() {
+			return (producerProperties, binding, destination) -> {
+				if (binding.equals("retryInBinder-in-0")) {
+					latch.countDown();
+				}
+				else if (binding.equals("retryInContainer-in-0")) {
+					latch.countDown();
+				}
+			};
+		}
 	}
 
 }