diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 44c732138..3dbaf0aa9 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -183,16 +183,16 @@ public class KafkaMessageChannelBinder
 		this.zookeeperConnect = zookeeperConnect;
 		this.brokers = brokers;
 		this.zkAddress = zkAddress;
-		if (headersToMap.length > 0) {
+		if (ObjectUtils.isEmpty(headersToMap)) {
+			this.headersToMap = BinderHeaders.STANDARD_HEADERS;
+		}
+		else {
 			String[] combinedHeadersToMap = Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0,
 					BinderHeaders.STANDARD_HEADERS.length + headersToMap.length);
 			System.arraycopy(headersToMap, 0, combinedHeadersToMap, BinderHeaders.STANDARD_HEADERS.length,
 					headersToMap.length);
 			this.headersToMap = combinedHeadersToMap;
 		}
-		else {
-			this.headersToMap = BinderHeaders.STANDARD_HEADERS;
-		}
 	}
 
 	String getZkAddress() {
@@ -793,7 +793,6 @@ public class KafkaMessageChannelBinder
 		}
 
 		@Override
-		@SuppressWarnings("unchecked")
 		protected Object handleRequestMessage(Message<?> requestMessage) {
 			if (HeaderMode.embeddedHeaders.equals(consumerProperties.getHeaderMode())) {
 				MessageValues messageValues = extractMessageValues(requestMessage);
diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
index 2a1a461bc..10def801e 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
@@ -32,7 +32,6 @@ import org.springframework.integration.codec.Codec;
 import org.springframework.integration.kafka.support.LoggingProducerListener;
 import org.springframework.integration.kafka.support.ProducerListener;
 import org.springframework.integration.kafka.support.ZookeeperConnect;
-import org.springframework.util.ObjectUtils;
 
 /**
  * @author David Turanski
@@ -71,10 +70,8 @@ public class KafkaBinderConfiguration {
 		String[] headers = kafkaBinderConfigurationProperties.getHeaders();
 		String kafkaConnectionString = kafkaBinderConfigurationProperties.getKafkaConnectionString();
 		String zkConnectionString = kafkaBinderConfigurationProperties.getZkConnectionString();
-		KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
-				new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString)
-				: new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString,
-				headers);
+		KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
+				zookeeperConnect(), kafkaConnectionString, zkConnectionString, headers);
 		kafkaMessageChannelBinder.setCodec(codec);
 		kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(kafkaBinderConfigurationProperties.getOffsetUpdateTimeWindow());
 		kafkaMessageChannelBinder.setOffsetUpdateCount(kafkaBinderConfigurationProperties.getOffsetUpdateCount());