Support headers in Kafka Streams binder

* Use content type header from the record for binder provided inbound deserialization by
  making use of the header support added in Kafka Streams. If there is a content type set
  on the incoming record, that will get precedence.
* Introduce a new Composite Serde class for providing non-native Spring Cloud Stream specific collection
  of Serde implemenations. This is needed in order for things like avro converters that interact with
  the Spring Cloud Stream schema registry server.
* Adding tests
* Polishing

Resolves #456, #469

pom.xml

@@ -129,6 +129,12 @@
 					</exclusion>
 				</exclusions>
 			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-schema</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+				<scope>test</scope>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 

spring-cloud-stream-binder-kafka-streams/pom.xml

@@ -13,6 +13,10 @@
 		<version>2.1.0.BUILD-SNAPSHOT</version>
 	</parent>
 
+	<properties>
+		<avro.version>1.8.2</avro.version>
+	</properties>
+
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
@@ -76,5 +80,40 @@
 			<classifier>test</classifier>
 			<scope>test</scope>
 		</dependency>
+		<!-- Following dependencies are only provided for testing and won't be packaged with the binder apps-->
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-schema</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.avro</groupId>
+			<artifactId>avro</artifactId>
+			<version>${avro.version}</version>
+			<scope>provided</scope>
+		</dependency>
 	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.avro</groupId>
+				<artifactId>avro-maven-plugin</artifactId>
+				<version>${avro.version}</version>
+				<executions>
+					<execution>
+						<phase>generate-sources</phase>
+						<goals>
+							<goal>schema</goal>
+							<goal>protocol</goal>
+							<goal>idl-protocol</goal>
+						</goals>
+						<configuration>
+							<sourceDirectory>src/test/resources/avro</sourceDirectory>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
 </project>

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java

@@ -36,6 +36,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.kafka.streams.serde.CompositeNonNativeSerde;
 import org.springframework.cloud.stream.binding.BindingService;
 import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
@@ -161,6 +162,11 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
 				KafkaStreamsBindingInformationCatalogue, binderConfigurationProperties);
 	}
 
+	@Bean
+	public CompositeNonNativeSerde compositeNonNativeSerde(CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		return new CompositeNonNativeSerde(compositeMessageConverterFactory);
+	}
+
 	@Bean
 	public KStreamBoundElementFactory kStreamBoundElementFactory(BindingServiceProperties bindingServiceProperties,
 																	KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) {

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java

@@ -21,6 +21,8 @@ import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.processor.Processor;
@@ -92,7 +94,7 @@ public class KafkaStreamsMessageConversionDelegate {
 	}
 
 	/**
-	 * Deserialize incoming {@link KStream} based on contentType.
+	 * Deserialize incoming {@link KStream} based on content type.
 	 *
 	 * @param valueClass on KStream value
 	 * @param bindingTarget inbound KStream target
@@ -101,6 +103,9 @@ public class KafkaStreamsMessageConversionDelegate {
 	@SuppressWarnings("unchecked")
 	public KStream deserializeOnInbound(Class<?> valueClass, KStream<?, ?> bindingTarget) {
 		MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
+		final PerRecordContentTypeHolder perRecordContentTypeHolder = new PerRecordContentTypeHolder();
+
+		resolvePerRecordContentType(bindingTarget, perRecordContentTypeHolder);
 
 		//Deserialize using a branching strategy
 		KStream<?, ?>[] branch = bindingTarget.branch(
@@ -113,16 +118,26 @@ public class KafkaStreamsMessageConversionDelegate {
 					if (o2 != null) {
 						if (valueClass.isAssignableFrom(o2.getClass())) {
 							keyValueThreadLocal.set(new KeyValue<>(o, o2));
-						} else if (o2 instanceof Message) {
-							if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) {
-								keyValueThreadLocal.set(new KeyValue<>(o, ((Message) o2).getPayload()));
-							} else {
-								convertAndSetMessage(o, valueClass, messageConverter, (Message) o2);
+						}
+						else if (o2 instanceof Message) {
+							Message<?> m1 = (Message) o2;
+							if (perRecordContentTypeHolder.contentType != null) {
+								m1 = MessageBuilder.fromMessage(m1).setHeader("contentType", perRecordContentTypeHolder.contentType).build();
 							}
-						} else if (o2 instanceof String || o2 instanceof byte[]) {
-							Message<?> message = MessageBuilder.withPayload(o2).build();
+
+							if (valueClass.isAssignableFrom(m1.getPayload().getClass())) {
+								keyValueThreadLocal.set(new KeyValue<>(o, m1.getPayload()));
+							}
+							else {
+								convertAndSetMessage(o, valueClass, messageConverter, m1);
+							}
+						}
+						else if (o2 instanceof String || o2 instanceof byte[]) {
+							Message<?> message = perRecordContentTypeHolder.contentType != null ? MessageBuilder.withPayload(o2)
+									.setHeader("contentType", perRecordContentTypeHolder.contentType).build() : MessageBuilder.withPayload(o2).build();
 							convertAndSetMessage(o, valueClass, messageConverter, message);
-						} else {
+						}
+						else {
 							keyValueThreadLocal.set(new KeyValue<>(o, o2));
 						}
 						isValidRecord = true;
@@ -150,6 +165,45 @@ public class KafkaStreamsMessageConversionDelegate {
 		});
 	}
 
+	private static class PerRecordContentTypeHolder {
+
+		String contentType;
+
+		void setContentType(String contentType) {
+			this.contentType = contentType;
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void resolvePerRecordContentType(KStream<?, ?> outboundBindTarget, PerRecordContentTypeHolder perRecordContentTypeHolder) {
+		outboundBindTarget.process(() -> new Processor() {
+
+			ProcessorContext context;
+
+			@Override
+			public void init(ProcessorContext context) {
+				this.context = context;
+			}
+
+			@Override
+			public void process(Object key, Object value) {
+				final Headers headers = context.headers();
+				final Iterable<Header> contentTypes = headers.headers("contentType");
+				if (contentTypes != null && contentTypes.iterator().hasNext()) {
+					final String contentType = new String(contentTypes.iterator().next().value());
+					//remove leading and trailing quotes
+					final String cleanContentType = StringUtils.replace(contentType, "\"", "");
+					perRecordContentTypeHolder.setContentType(cleanContentType);
+				}
+			}
+
+			@Override
+			public void close() {
+
+			}
+		});
+	}
+
 	private void convertAndSetMessage(Object o, Class<?> valueClass, MessageConverter messageConverter, Message<?> msg) {
 		Object messageConverted = messageConverter.fromMessage(msg, valueClass);
 		if (messageConverted == null) {

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java

@@ -0,0 +1,210 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams.serde;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * A {@link Serde} implementation that wraps the list of {@link MessageConverter}s
+ * from {@link CompositeMessageConverterFactory}.
+ *
+ * The primary motivation for this class is to provide an avro based {@link Serde} that is
+ * compatible with the schema registry that Spring Cloud Stream provides. When using the
+ * schema registry support from Spring Cloud Stream in a Kafka Streams binder based application,
+ * the applications can deserialize the incoming Kafka Streams records using the built in
+ * Avro {@link MessageConverter}. However, this same message conversion approach will not work
+ * downstream in other operations in the topology for Kafka Streams as some of them needs a
+ * {@link Serde} instance that can talk to the Spring Cloud Stream provided Schema Registry.
+ * This implementation will solve that problem.
+ *
+ * Only Avro and JSON based converters are exposed as binder provided {@link Serde} implementations currently.
+ *
+ * Users of this class must call the {@link CompositeNonNativeSerde#configure(Map, boolean)} method
+ * to configure the {@link Serde} object. At the very least the configuration map must include a key
+ * called "valueClass" to indicate the type of the target object for deserialization. If any other
+ * content type other than JSON is needed (only Avro is available now other than JSON), that needs
+ * to be included in the configuration map with the key "contentType". For example,
+ *
+ * <pre class="code">
+ * Map<String, Object> config = new HashMap<>();
+ * config.put("valueClass", Foo.class);
+ * config.put("contentType", "application/avro");
+ * </pre>
+ *
+ * Then use the above map when calling the configure method.
+ *
+ * This class is only intended to be used when writing a Spring Cloud Stream Kafka Streams application
+ * that uses Spring Cloud Stream schema registry for schema evolution.
+ *
+ * An instance of this class is provided as a bean by the binder configuration and typically the applications
+ * can autowire that bean. This is the expected usage pattern of this class.
+ *
+ * @author Soby Chacko
+ */
+public class CompositeNonNativeSerde<T> implements Serde<T> {
+
+	private static final String CONTENT_TYPE_HEADER = "contentType";
+
+	private static final String VALUE_CLASS_HEADER = "valueClass";
+
+	private static final String AVRO_FORMAT = "avro";
+
+	private static final MimeType DEFAULT_AVRO_MIME_TYPE = new MimeType("application", "*+" + AVRO_FORMAT);
+
+	private final CompositeNonNativeDeserializer<T> compositeNonNativeDeserializer;
+
+	private final CompositeNonNativeSerializer<T> compositeNonNativeSerializer;
+
+	public CompositeNonNativeSerde(CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		this.compositeNonNativeDeserializer = new CompositeNonNativeDeserializer<>(compositeMessageConverterFactory);
+		this.compositeNonNativeSerializer = new CompositeNonNativeSerializer<>(compositeMessageConverterFactory);
+	}
+
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		this.compositeNonNativeDeserializer.configure(configs, isKey);
+		this.compositeNonNativeSerializer.configure(configs, isKey);
+	}
+
+	@Override
+	public void close() {
+		//No-op
+	}
+
+	@Override
+	public Serializer<T> serializer() {
+		return this.compositeNonNativeSerializer;
+	}
+
+	@Override
+	public Deserializer<T> deserializer() {
+		return this.compositeNonNativeDeserializer;
+	}
+
+	private static MimeType resolveMimeType(Map<String, ?> configs) {
+		if (configs.containsKey(CONTENT_TYPE_HEADER)){
+			String contentType = (String)configs.get(CONTENT_TYPE_HEADER);
+			if (DEFAULT_AVRO_MIME_TYPE.equals(MimeTypeUtils.parseMimeType(contentType))) {
+				return DEFAULT_AVRO_MIME_TYPE;
+			}
+			else if(contentType.contains("avro")) {
+				return MimeTypeUtils.parseMimeType("application/avro");
+			}
+			else {
+				return new MimeType("application", "json", StandardCharsets.UTF_8);
+			}
+		}
+		else {
+			return new MimeType("application", "json", StandardCharsets.UTF_8);
+		}
+	}
+
+	/**
+	 * Custom {@link Deserializer} that uses the {@link CompositeMessageConverterFactory}.
+	 *
+	 * @param <U> Parameterized target type for deserialization
+	 */
+	private static class CompositeNonNativeDeserializer<U> implements Deserializer<U> {
+
+		private final MessageConverter messageConverter;
+
+		private MimeType mimeType;
+
+		private Class<?> valueClass;
+
+		CompositeNonNativeDeserializer(CompositeMessageConverterFactory compositeMessageConverterFactory) {
+			this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
+		}
+
+		@Override
+		public void configure(Map<String, ?> configs, boolean isKey) {
+			Assert.isTrue(configs.containsKey(VALUE_CLASS_HEADER), "Deserializers must provide a configuration for valueClass.");
+			final Object valueClass = configs.get(VALUE_CLASS_HEADER);
+			Assert.isTrue(valueClass instanceof Class, "Deserializers must provide a valid value for valueClass.");
+			this.valueClass = (Class<?>) valueClass;
+			this.mimeType = resolveMimeType(configs);
+		}
+
+		@SuppressWarnings("unchecked")
+		@Override
+		public U deserialize(String topic, byte[] data) {
+			Message<?> message = MessageBuilder.withPayload(data)
+					.setHeader(CONTENT_TYPE_HEADER, this.mimeType.toString()).build();
+			U messageConverted = (U)messageConverter.fromMessage(message, this.valueClass);
+			if (messageConverted == null) {
+				throw new IllegalStateException("Deserialization failed.");
+			}
+			return messageConverted;
+		}
+
+		@Override
+		public void close() {
+			//No-op
+		}
+	}
+
+	/**
+	 * Custom {@link Serializer} that uses the {@link CompositeMessageConverterFactory}.
+	 *
+	 * @param <V>  Parameterized type for serialization
+	 */
+	private static class CompositeNonNativeSerializer<V> implements Serializer<V> {
+
+		private final MessageConverter messageConverter;
+		private MimeType mimeType;
+
+		CompositeNonNativeSerializer(CompositeMessageConverterFactory compositeMessageConverterFactory) {
+			this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
+		}
+
+		@Override
+		public void configure(Map<String, ?> configs, boolean isKey) {
+			this.mimeType = resolveMimeType(configs);
+		}
+
+		@Override
+		public byte[] serialize(String topic, V data) {
+			Message<?> message = MessageBuilder.withPayload(data).build();
+			Map<String, Object> headers = new HashMap<>(message.getHeaders());
+			headers.put(MessageHeaders.CONTENT_TYPE, this.mimeType.toString());
+			MessageHeaders messageHeaders = new MessageHeaders(headers);
+			final Object payload = messageConverter.toMessage(message.getPayload(),
+					messageHeaders).getPayload();
+			return (byte[])payload;
+		}
+
+		@Override
+		public void close() {
+			//No-op
+		}
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2017-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams.integration;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+
+import com.example.Sensor;
+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.ByteArrayDeserializer;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.annotation.StreamMessageConverter;
+import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
+import org.springframework.cloud.stream.binder.kafka.streams.integration.utils.TestAvroSerializer;
+import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+public class PerRecordAvroContentTypeTests {
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, "received-sensors");
+
+	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+
+	private static Consumer<String, byte[]> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("avro-ct-test", "false", embeddedKafka);
+
+		//Receive the data as byte[]
+		consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, byte[]> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "received-sensors");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	@Test
+	public void testPerRecordAvroConentTypeAndVerifySerialization() throws Exception {
+		SpringApplication app = new SpringApplication(SensorCountAvroApplication.class);
+		app.setWebApplicationType(WebApplicationType.NONE);
+
+		try (ConfigurableApplicationContext ignored = app.run("--server.port=0",
+				"--spring.jmx.enabled=false",
+				"--spring.cloud.stream.bindings.input.destination=sensors",
+				"--spring.cloud.stream.bindings.output.destination=received-sensors",
+				"--spring.cloud.stream.bindings.output.contentType=application/avro",
+				"--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=per-record-avro-contentType-test",
+				"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
+
+			Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+			//Use a custom avro test serializer
+			senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, TestAvroSerializer.class);
+			DefaultKafkaProducerFactory<Integer, Sensor> pf = new DefaultKafkaProducerFactory<>(senderProps);
+			try {
+				KafkaTemplate<Integer, Sensor> template = new KafkaTemplate<>(pf, true);
+
+				Random random = new Random();
+				Sensor sensor = new Sensor();
+				sensor.setId(UUID.randomUUID().toString() + "-v1");
+				sensor.setAcceleration(random.nextFloat() * 10);
+				sensor.setVelocity(random.nextFloat() * 100);
+				sensor.setTemperature(random.nextFloat() * 50);
+				//Send with avro content type set.
+				Message<?> message = MessageBuilder.withPayload(sensor)
+						.setHeader("contentType", "application/avro")
+						.build();
+				template.setDefaultTopic("sensors");
+				template.send(message);
+
+				//Serialized byte[] ^^ is received by the binding process and deserialzed it using avro converter.
+				//Then finally, the data will be output to a return topic as byte[] (using the same avro converter).
+
+				//Receive the byte[] from return topic
+				ConsumerRecord<String, byte[]> cr = KafkaTestUtils.getSingleRecord(consumer, "received-sensors");
+				final byte[] value = cr.value();
+
+				//Convert the byte[] received back to avro object and verify that it is the same as the one we sent ^^.
+				AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter();
+
+				Message<?> receivedMessage = MessageBuilder.withPayload(value)
+						.setHeader("contentType", MimeTypeUtils.parseMimeType("application/avro")).build();
+				Sensor messageConverted = (Sensor)avroSchemaMessageConverter.fromMessage(receivedMessage, Sensor.class);
+				assertThat(messageConverted).isEqualTo(sensor);
+			}
+			finally {
+				pf.destroy();
+			}
+		}
+	}
+
+	@EnableBinding(KafkaStreamsProcessor.class)
+	@EnableAutoConfiguration
+	static class SensorCountAvroApplication {
+
+		@StreamListener
+		@SendTo("output")
+		public KStream<?, Sensor> process(@Input("input") KStream<Object, Sensor> input) {
+			//return the same Sensor object unchanged so that we can do test verifications
+			return input.map(KeyValue::new);
+		}
+
+		@Bean
+		@StreamMessageConverter
+		public MessageConverter sensorMessageConverter() throws IOException {
+			return new AvroSchemaMessageConverter();
+		}
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams.integration.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.common.serialization.Serializer;
+
+import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Custom avro serializer intended to be used for testing only.
+ *
+ * @author Soby Chacko
+ *
+ * @param <S> Target type to serialize
+ */
+public class TestAvroSerializer<S> implements Serializer<S> {
+
+	public TestAvroSerializer() {}
+
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+
+	}
+
+	@Override
+	public byte[] serialize(String topic, S data) {
+		AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter();
+		Message<?> message = MessageBuilder.withPayload(data).build();
+		Map<String, Object> headers = new HashMap<>(message.getHeaders());
+		headers.put(MessageHeaders.CONTENT_TYPE, "application/avro");
+		MessageHeaders messageHeaders = new MessageHeaders(headers);
+		final Object payload = avroSchemaMessageConverter.toMessage(message.getPayload(),
+				messageHeaders).getPayload();
+		return (byte[])payload;
+	}
+
+	@Override
+	public void close() {
+
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerdeTest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams.serde;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+
+import com.example.Sensor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Test;
+
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Refer {@link CompositeNonNativeSerde} for motivations.
+ *
+ * @author Soby Chacko
+ */
+public class CompositeNonNativeSerdeTest {
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testCompositeNonNativeSerdeUsingAvroContentType(){
+		Random random = new Random();
+		Sensor sensor = new Sensor();
+		sensor.setId(UUID.randomUUID().toString() + "-v1");
+		sensor.setAcceleration(random.nextFloat() * 10);
+		sensor.setVelocity(random.nextFloat() * 100);
+		sensor.setTemperature(random.nextFloat() * 50);
+
+		List<MessageConverter> messageConverters = new ArrayList<>();
+		messageConverters.add(new AvroSchemaMessageConverter());
+		CompositeMessageConverterFactory compositeMessageConverterFactory =
+				new CompositeMessageConverterFactory(messageConverters, new ObjectMapper());
+		CompositeNonNativeSerde compositeNonNativeSerde = new CompositeNonNativeSerde(compositeMessageConverterFactory);
+
+		Map<String, Object> configs = new HashMap<>();
+		configs.put("valueClass", Sensor.class);
+		configs.put("contentType", "application/avro");
+		compositeNonNativeSerde.configure(configs, false);
+		final byte[] serialized = compositeNonNativeSerde.serializer().serialize(null, sensor);
+
+		final Object deserialized = compositeNonNativeSerde.deserializer().deserialize(null, serialized);
+
+		assertThat(deserialized).isEqualTo(sensor);
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/test/resources/avro/sensor.avsc

@@ -0,0 +1,11 @@
+{
+  "namespace" : "com.example",
+  "type" : "record",
+  "name" : "Sensor",
+  "fields" : [
+    {"name":"id","type":"string"},
+    {"name":"temperature", "type":"float", "default":0.0},
+    {"name":"acceleration", "type":"float","default":0.0},
+    {"name":"velocity","type":"float","default":0.0}
+  ]
+}