From b0c4f0cfcdba1b751791b138eb3d3a59d457d1a7 Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Tue, 14 Nov 2017 16:42:36 -0500
Subject: [PATCH] Addressing PR review comments

Resolves #258
---
 .../cloud/stream/binder/kafka/KafkaMessageChannelBinder.java | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 0f44e3c1c..e1a65adf7 100644
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -439,9 +439,10 @@ public class KafkaMessageChannelBinder extends
 				final ConsumerRecord<?, ?> record = message.getHeaders()
 						.get(KafkaHeaders.RAW_DATA, ConsumerRecord.class);
 
-				if (this.transactionManager == null && extendedConsumerProperties.isUseNativeDecoding()) {
+				if (extendedConsumerProperties.isUseNativeDecoding()) {
 					if (record != null) {
-						Map<String, String> configuration = dlqProducerProperties.getConfiguration();
+						Map<String, String> configuration = this.transactionManager == null ? dlqProducerProperties.getConfiguration()
+								: this.configurationProperties.getTransaction().getProducer().getConfiguration();
 						if (record.key() != null && !record.key().getClass().isInstance(byte[].class)) {
 							ensureDlqMessageCanBeProperlySerialized(
 									configuration,