diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java
index 5fc97ee8f..5cde2dd0c 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java
@@ -31,10 +31,14 @@ import java.util.concurrent.atomic.AtomicReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
+
 import org.I0Itec.zkclient.ZkClient;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -96,6 +100,7 @@ import static org.assertj.core.api.Assertions.fail;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
+
 /**
  * @author Soby Chacko
  * @author Ilayaperumal Gopinathan
@@ -1454,6 +1459,59 @@ public abstract class AbstractKafkaBinderTests extends
 	}
 
 	//TODO: We need to evaluate this test as built in serialization has different meaning in 2.0
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload() throws Exception {
+		Binding<?> producerBinding = null;
+		Binding<?> consumerBinding = null;
+		try {
+			byte[] testPayload = new byte[1];
+			Message<?> message = MessageBuilder.withPayload(testPayload)
+					.setHeader(MessageHeaders.CONTENT_TYPE, "something/funky")
+					.build();
+			SubscribableChannel moduleOutputChannel = new DirectChannel();
+			String testTopicName = "existing" + System.currentTimeMillis();
+			KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+			final ZkClient zkClient;
+			zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
+					configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
+					ZKStringSerializer$.MODULE$);
+			final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+			invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties());
+			configurationProperties.setAutoAddPartitions(true);
+			Binder binder = getBinder(configurationProperties);
+			QueueChannel moduleInputChannel = new QueueChannel();
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+			producerProperties.setUseNativeEncoding(true);
+			producerProperties.getExtension()
+					.getConfiguration()
+					.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+			producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties);
+			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+			consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+			consumerProperties.getExtension()
+					.getConfiguration()
+					.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+			consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
+			// Let the consumer actually bind to the producer before sending a msg
+			binderBindUnbindLatency();
+			moduleOutputChannel.send(message);
+			Message<?> inbound = receive(moduleInputChannel, 500);
+			assertThat(inbound).isNotNull();
+			assertThat(inbound.getPayload()).isEqualTo(new byte[1]);
+			assertThat(inbound.getHeaders()).containsKey("contentType");
+			assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("something/funky");
+		}
+		finally {
+			if (producerBinding != null) {
+				producerBinding.unbind();
+			}
+			if (consumerBinding != null) {
+				consumerBinding.unbind();
+			}
+		}
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testBuiltinSerialization() throws Exception {