Release 1.3.0.M2

* Documentation for KStream binder

Fix #160

* Addressing PR review comments

* Addressing PR review comments

* Addressing PR review comments

pom.xml

@@ -2,7 +2,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-	<version>1.3.0.M1</version>
+	<version>1.3.0.M2</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
@@ -15,7 +15,8 @@
 		<kafka.version>0.10.1.1</kafka.version>
 		<spring-kafka.version>1.1.6.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>2.1.1.RELEASE</spring-integration-kafka.version>
-		<spring-cloud-stream.version>1.3.0.M1</spring-cloud-stream.version>
+		<spring-cloud-stream.version>1.3.0.M2</spring-cloud-stream.version>
+		<spring-cloud-build.version>1.3.3.RELEASE</spring-cloud-build.version>
 	</properties>
 	<modules>
 		<module>spring-cloud-stream-binder-kafka</module>
@@ -24,6 +25,7 @@
 		<module>spring-cloud-stream-binder-kafka-0.10.1-test</module>
 		<module>spring-cloud-stream-binder-kafka-0.10.2-test</module>
 		<module>spring-cloud-stream-binder-kafka-core</module>
+		<module>spring-cloud-stream-binder-kstream</module>
     </modules>
 
 	<dependencyManagement>
@@ -100,6 +102,17 @@
 				<classifier>test</classifier>
 				<version>${kafka.version}</version>
 			</dependency>
+			<dependency>
+				<groupId>org.apache.kafka</groupId>
+				<artifactId>kafka-streams</artifactId>
+				<version>${kafka.version}</version>
+				<exclusions>
+					<exclusion>
+						<groupId>org.slf4j</groupId>
+						<artifactId>slf4j-log4j12</artifactId>
+					</exclusion>
+				</exclusions>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 
@@ -147,7 +160,7 @@
 					<dependency>
 						<groupId>org.springframework.cloud</groupId>
 						<artifactId>spring-cloud-build-tools</artifactId>
-						<version>${project.parent.version}</version>
+						<version>${spring-cloud-build.version}</version>
 					</dependency>
 				</dependencies>
 				<executions>

spring-cloud-starter-stream-kafka/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.3.0.M1</version>
+		<version>1.3.0.M2</version>
 	</parent>
 	<artifactId>spring-cloud-starter-stream-kafka</artifactId>
 	<description>Spring Cloud Starter Stream Kafka</description>

spring-cloud-stream-binder-kafka-0.10.1-test/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.3.0.M1</version>
+        <version>1.3.0.M2</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-0.10.1-test</artifactId>
     <description>Spring Cloud Stream Kafka Binder 0.10.1 Tests</description>

spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.3.0.M1</version>
+        <version>1.3.0.M2</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-0.10.2-test</artifactId>
     <description>Spring Cloud Stream Kafka Binder 0.10.2 Tests</description>

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

@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.3.0.M1</version>
+        <version>1.3.0.M2</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
     <description>Spring Cloud Stream Kafka Binder Core</description>

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

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.3.0.M1</version>
+        <version>1.3.0.M2</version>
     </parent>
 
     <artifactId>spring-cloud-stream-binder-kafka-docs</artifactId>

spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc

@@ -2,6 +2,7 @@
 --
 This guide describes the Apache Kafka implementation of the Spring Cloud Stream Binder.
 It contains information about its design, usage and configuration options, as well as information on how the Stream Cloud Stream concepts map into Apache Kafka specific constructs.
+In addition, this guide also explains the Kafka Streams binding capabilities of Spring Cloud Stream.
 --
 
 == Usage
@@ -429,4 +430,98 @@ On the other hand, if auto topic creation is disabled on the server, then care m
 
 If you want to have full control over how partitions are allocated, then leave the default settings as they are, i.e. do not exclude the kafka broker jar and ensure that `spring.cloud.stream.kafka.binder.autoCreateTopics` is set to `true`, which is the default.
 
+== Kafka Streams Binding Capabilities of Spring Cloud Stream
+
+Spring Cloud Stream Kafka support also includes a binder specifically designed for Kafka Streams binding.
+Using this binder, applications can be written that leverage the Kafka Streams API.
+For more information on Kafka Streams, see https://kafka.apache.org/documentation/streams/developer-guide[Kafka Streams API Developer Manual]
+
+Kafka Streams support in Spring Cloud Stream is based on the foundations provided by the Spring Kafka project. For details on that support, see http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafaka Streams Support in Spring Kafka].
+
+Here are the maven coordinates for the Spring Cloud Stream KStream binder artifact.
+
+[source,xml]
+----
+<dependency>
+  <groupId>org.springframework.cloud</groupId>
+  <artifactId>spring-cloud-stream-binder-kstream</artifactId>
+</dependency>
+----
+
+In addition to leveraging the Spring Cloud Stream programming model which is based on Spring Boot, one of the main other benefits that the KStream binder provides is the fact that it avoids the boilerplate configuration that one needs to write when using the Kafka Streams API directly.
+High level streams DSL provided through the Kafka Streams API can be used through Spring Cloud Stream in the current support.
+
+=== Usage example of high level streams DSL
+
+This application will listen from a Kafka topic and write the word count for each unique word that it sees in a 5 seconds time window.
+
+[source]
+----
+@SpringBootApplication
+@EnableBinding(KStreamProcessor.class)
+public class WordCountProcessorApplication {
+
+	@StreamListener("input")
+	@SendTo("output")
+	public KStream<?, String> process(KStream<?, String> input) {
+		return input
+				.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
+				.map((key, word) -> new KeyValue<>(word, word))
+				.groupByKey(Serdes.String(), Serdes.String())
+				.count(TimeWindows.of(5000), "store-name")
+				.toStream()
+				.map((w, c) -> new KeyValue<>(null, "Count for " + w.key() + ": " + c));
+	}
+
+	public static void main(String[] args) {
+		SpringApplication.run(WordCountProcessorApplication.class, args);
+	}
+----
+
+If you build it as Spring Boot runnable fat jar, you can run the above example in the following way:
+
+[source]
+----
+java -jar uber.jar  --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts
+----
+
+This means that the application will listen from the incoming Kafka topic words and write to the output topic counts.
+
+Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are bound as KStream objects.
+As one may observe, the developer can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor rather than setting up the streams specific configuration required by the Kafka Streams infrastructure.
+All those boilerplate is handled by Spring Cloud Stream behind the scenes.
+
+=== Support for interactive queries
+
+If access to the `KafkaStreams` is needed for interactive queries, the internal `KafkaStreams` instance can be accessed via `KStreamBuilderFactoryBean.getKafkaStreams()`.
+You can autowire the `KStreamBuilderFactoryBean` instance provided by the KStream binder. Then you can get `KafkaStreams` instance from it and retrieve the underlying store, execute queries on it, etc.
+
+=== Kafka Streams properties
+
+configuration::
+  Map with a key/value pair containing properties pertaining to Kafka Streams API.
+  This property must be prefixed with `spring.cloud.stream.kstream.binder.`.
+
+ Following are some examples of using this property.
+
+[source]
+----
+spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000
+----
+
+  For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs.
+
+There can also be binding specific properties.
+
+For instance, you can use a different Serde for your input or output destination.
+
+[source]
+----
+spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde
+spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde
+----
+
+
 

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

@@ -10,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.3.0.M1</version>
+		<version>1.3.0.M2</version>
 	</parent>
 
 	<dependencies>

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

@@ -282,10 +282,6 @@ public class KafkaMessageChannelBinder extends
 			this.logger.debug(
 					"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
 		}
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug(
-					"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
-		}
 		final KafkaMessageDrivenChannelAdapter<?, ?> kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(
 				messageListenerContainer);
 		kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());

spring-cloud-stream-binder-kafka/src/test/resources/logback.xml

@@ -11,4 +11,4 @@
     <root level="WARN">
         <appender-ref ref="stdout"/>
     </root>
-</configuration>
+</configuration>

spring-cloud-stream-binder-kstream/pom.xml

@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-stream-binder-kstream</artifactId>
+	<packaging>jar</packaging>
+	<name>spring-cloud-stream-binder-kstream</name>
+	<description>Kafka Streams Binder Implementation</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
+		<version>1.3.0.M2</version>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-codec</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-autoconfigure</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.11</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-streams</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.11</artifactId>
+			<classifier>test</classifier>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+
+import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.DefaultBinding;
+import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binder.HeaderMode;
+import org.springframework.cloud.stream.binder.MessageValues;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
+import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamBinder extends
+		AbstractBinder<KStream<Object, Object>, ExtendedConsumerProperties<KStreamConsumerProperties>, ExtendedProducerProperties<KStreamProducerProperties>>
+		implements ExtendedPropertiesBinder<KStream<Object, Object>, KStreamConsumerProperties, KStreamProducerProperties> {
+
+	private String[] headers;
+
+	private final KafkaTopicProvisioner kafkaTopicProvisioner;
+
+	private final KStreamExtendedBindingProperties kStreamExtendedBindingProperties;
+
+	private final StreamsConfig streamsConfig;
+
+	public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner,
+			KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
+		this.headers = EmbeddedHeaderUtils.headersToEmbed(binderConfigurationProperties.getHeaders());
+		this.kafkaTopicProvisioner = kafkaTopicProvisioner;
+		this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
+		this.streamsConfig = streamsConfig;
+	}
+
+	@Override
+	protected Binding<KStream<Object, Object>> doBindConsumer(String name, String group,
+			KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<KafkaConsumerProperties>(
+				new KafkaConsumerProperties());
+		this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties);
+		return new DefaultBinding<>(name, group, inputTarget, null);
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected Binding<KStream<Object, Object>> doBindProducer(String name, KStream<Object, Object> outboundBindTarget,
+			ExtendedProducerProperties<KStreamProducerProperties> properties) {
+		ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<KafkaProducerProperties>(
+				new KafkaProducerProperties());
+		this.kafkaTopicProvisioner.provisionProducerDestination(name , extendedProducerProperties);
+		if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
+			outboundBindTarget = outboundBindTarget.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
+				@Override
+				public KeyValue<Object, Object> apply(Object k, Object v) {
+					if (v instanceof Message) {
+						try {
+							return new KeyValue<>(k, (Object)KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
+						}
+						catch (Exception e) {
+							throw new IllegalArgumentException(e);
+						}
+					}
+					else {
+						throw new IllegalArgumentException("Wrong type of message " + v);
+					}
+				}
+			});
+		}
+		else {
+			if (!properties.isUseNativeEncoding()) {
+				outboundBindTarget = outboundBindTarget
+						.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
+							@Override
+							public KeyValue<Object, Object> apply(Object k, Object v) {
+								return KeyValue.pair(k, (Object)KStreamBinder.this.serializePayloadIfNecessary((Message<?>) v));
+							}
+						});
+			}
+			else {
+				outboundBindTarget = outboundBindTarget
+						.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
+							@Override
+							public KeyValue<Object, Object> apply(Object k, Object v) {
+								return KeyValue.pair(k, ((Message<?>) v).getPayload());
+							}
+						});
+			}
+		}
+		if (!properties.isUseNativeEncoding() || StringUtils.hasText(properties.getExtension().getKeySerde()) || StringUtils.hasText(properties.getExtension().getValueSerde())) {
+			Serde<?> keySerde = Serdes.ByteArray();
+			Serde<?> valueSerde = Serdes.ByteArray();
+			try {
+				if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
+					keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
+					if (keySerde instanceof Configurable) {
+						((Configurable) keySerde).configure(streamsConfig.originals());
+					}
+				}
+			}
+			catch (ClassNotFoundException e) {
+				throw new IllegalStateException("Serde class not found: ", e);
+			}
+			try {
+				if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
+					valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
+					if (valueSerde instanceof Configurable) {
+						((Configurable) valueSerde).configure(streamsConfig.originals());
+					}
+				}
+			}
+			catch (ClassNotFoundException e) {
+				throw new IllegalStateException("Serde class not found: ", e);
+			}
+			outboundBindTarget.to((Serde<Object>) keySerde, (Serde<Object>) valueSerde, name);
+		}
+		else {
+			outboundBindTarget.to(name);
+		}
+		return new DefaultBinding<>(name, null, outboundBindTarget, null);
+	}
+
+	private byte[] serializeAndEmbedHeadersIfApplicable(Message<?> message) throws Exception {
+		MessageValues transformed = serializePayloadIfNecessary(message);
+		byte[] payload;
+
+		Object contentType = transformed.get(MessageHeaders.CONTENT_TYPE);
+		// transform content type headers to String, so that they can be properly embedded
+		// in JSON
+		if (contentType instanceof MimeType) {
+			transformed.put(MessageHeaders.CONTENT_TYPE, contentType.toString());
+		}
+		Object originalContentType = transformed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
+		if (originalContentType instanceof MimeType) {
+			transformed.put(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, originalContentType.toString());
+		}
+		payload = EmbeddedHeaderUtils.embedHeaders(transformed, headers);
+		return payload;
+	}
+
+	@Override
+	public KStreamConsumerProperties getExtendedConsumerProperties(String channelName) {
+		return this.kStreamExtendedBindingProperties.getExtendedConsumerProperties(channelName);
+	}
+
+	@Override
+	public KStreamProducerProperties getExtendedProducerProperties(String channelName) {
+		return this.kStreamExtendedBindingProperties.getExtendedProducerProperties(channelName);
+	}
+
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils;
+import org.springframework.cloud.stream.binder.HeaderMode;
+import org.springframework.cloud.stream.binder.MessageSerializationUtils;
+import org.springframework.cloud.stream.binder.MessageValues;
+import org.springframework.cloud.stream.binder.StringConvertingContentTypeResolver;
+import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.integration.codec.Codec;
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.Message;
+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.StringUtils;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> {
+
+	private final KStreamBuilder kStreamBuilder;
+
+	private final BindingServiceProperties bindingServiceProperties;
+
+	private volatile Codec codec;
+
+	private final StringConvertingContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
+
+	private volatile Map<String, Class<?>> payloadTypeCache = new ConcurrentHashMap<>();
+
+	private CompositeMessageConverterFactory compositeMessageConverterFactory;
+
+	public KStreamBoundElementFactory(KStreamBuilder streamBuilder, BindingServiceProperties bindingServiceProperties,
+			Codec codec, CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		super(KStream.class);
+		this.bindingServiceProperties = bindingServiceProperties;
+		this.kStreamBuilder = streamBuilder;
+		this.codec = codec;
+		this.compositeMessageConverterFactory = compositeMessageConverterFactory;
+	}
+
+	@Override
+	public KStream createInput(String name) {
+		KStream<Object, Object> stream = kStreamBuilder.stream(bindingServiceProperties.getBindingDestination(name));
+		ConsumerProperties properties = bindingServiceProperties.getConsumerProperties(name);
+		if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
+
+			stream = stream.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
+				@Override
+				public KeyValue<Object, Object> apply(Object key, Object value) {
+					if (!(value instanceof byte[])) {
+						return new KeyValue<>(key, value);
+					}
+					try {
+						MessageValues messageValues = EmbeddedHeaderUtils
+								.extractHeaders(MessageBuilder.withPayload((byte[]) value).build(), true);
+						messageValues = deserializePayloadIfNecessary(messageValues);
+						return new KeyValue<Object, Object>(null, messageValues.toMessage());
+					}
+					catch (Exception e) {
+						throw new IllegalArgumentException(e);
+					}
+				}
+			});
+		}
+		return stream;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public KStream createOutput(final String name) {
+		BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
+		String contentType = bindingProperties.getContentType();
+		MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory
+				.getMessageConverterForType(MimeType.valueOf(contentType)) : null;
+		KStreamWrapperHandler handler = new KStreamWrapperHandler(messageConverter);
+		ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class);
+		proxyFactory.addAdvice(handler);
+		return (KStream) proxyFactory.getProxy();
+	}
+
+	private MessageValues deserializePayloadIfNecessary(MessageValues messageValues) {
+		return MessageSerializationUtils.deserializePayload(messageValues, this.contentTypeResolver, this.codec);
+	}
+
+	interface KStreamWrapper {
+
+		void wrap(KStream<Object, Object> delegate);
+	}
+
+	static class KStreamWrapperHandler implements KStreamWrapper, MethodInterceptor {
+
+		private KStream<Object, Object> delegate;
+
+		private final MessageConverter messageConverter;
+
+		public KStreamWrapperHandler(MessageConverter messageConverter) {
+			this.messageConverter = messageConverter;
+		}
+
+		public void wrap(KStream<Object, Object> delegate) {
+			Assert.notNull(delegate, "delegate cannot be null");
+			Assert.isNull(this.delegate, "delegate already set to " + this.delegate);
+			if (messageConverter != null) {
+				KeyValueMapper<Object, Object, KeyValue<Object, Object>> keyValueMapper = new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
+					@Override
+					public KeyValue<Object, Object> apply(Object k, Object v) {
+						Message<?> message = (Message<?>) v;
+						return new KeyValue<Object, Object>(k,
+								messageConverter.toMessage(message.getPayload(),
+										new MutableMessageHeaders(((Message<?>) v).getHeaders())));
+					}
+				};
+				delegate = delegate.map(keyValueMapper);
+			}
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Object invoke(MethodInvocation methodInvocation) throws Throwable {
+			if (methodInvocation.getMethod().getDeclaringClass().equals(KStream.class)) {
+				Assert.notNull(delegate, "Trying to invoke " + methodInvocation
+						.getMethod() + "  but no delegate has been set.");
+				return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments());
+			}
+			else if (methodInvocation.getMethod().getDeclaringClass().equals(KStreamWrapper.class)) {
+				return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments());
+			}
+			else {
+				throw new IllegalStateException("Only KStream method invocations are permitted");
+			}
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+
+import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * @author Marius Bogoevici
+ * @author Soby Chacko
+ */
+public class KStreamListenerParameterAdapter implements StreamListenerParameterAdapter<KStream<?,?>, KStream<?, ?>> {
+
+	private final MessageConverter messageConverter;
+
+	public KStreamListenerParameterAdapter(MessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
+	@Override
+	public boolean supports(Class bindingTargetType, MethodParameter methodParameter) {
+		return KStream.class.isAssignableFrom(bindingTargetType)
+				&& KStream.class.isAssignableFrom(methodParameter.getParameterType());
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public KStream adapt(KStream<?, ?> bindingTarget, MethodParameter parameter) {
+		ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
+		final Class<?> valueClass = (resolvableType.getGeneric(1).getRawClass() != null)
+				? (resolvableType.getGeneric(1).getRawClass()) : Object.class;
+
+		return bindingTarget.map(new KeyValueMapper() {
+			@Override
+			public Object apply(Object o, Object o2) {
+				if (valueClass.isAssignableFrom(o2.getClass())) {
+					return new KeyValue<>(o, o2);
+				}
+				else if (o2 instanceof Message) {
+					return new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass));
+				}
+				else if(o2 instanceof String || o2 instanceof byte[]) {
+					Message<Object> message = MessageBuilder.withPayload(o2).build();
+					return new KeyValue<>(o, messageConverter.fromMessage(message, valueClass));
+				}
+				else {
+					return new KeyValue<>(o, o2);
+				}
+			}
+		});
+	}
+
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+
+import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamStreamListenerResultAdapter implements StreamListenerResultAdapter<KStream, KStreamBoundElementFactory.KStreamWrapper> {
+
+	@Override
+	public boolean supports(Class<?> resultType, Class<?> boundElement) {
+		return KStream.class.isAssignableFrom(resultType) && KStream.class.isAssignableFrom(boundElement);
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public Closeable adapt(KStream streamListenerResult, KStreamBoundElementFactory.KStreamWrapper boundElement) {
+		boundElement.wrap(streamListenerResult.map(new KeyValueMapper() {
+			@Override
+			public Object apply(Object k, Object v) {
+				if (v instanceof Message<?>) {
+					return new KeyValue<>(k, v);
+				}
+				else {
+					return new KeyValue<>(k, MessageBuilder.withPayload(v).build());
+				}
+			}
+		}));
+		return new NoOpCloseable();
+	}
+
+	private static final class NoOpCloseable implements Closeable {
+
+		@Override
+		public void close() throws IOException {
+
+		}
+
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/annotations/KStreamProcessor.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 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.kstream.annotations;
+
+import org.apache.kafka.streams.kstream.KStream;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.Output;
+
+/**
+ * @author Marius Bogoevici
+ */
+public interface KStreamProcessor {
+
+	@Input("input")
+	KStream<?, ?> input();
+
+	@Output("output")
+	KStream<?, ?> output();
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.streams.StreamsConfig;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kstream.KStreamBinder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * @author Marius Bogoevici
+ */
+@Configuration
+@EnableConfigurationProperties(KStreamExtendedBindingProperties.class)
+public class KStreamBinderConfiguration {
+
+	@Autowired(required = false)
+	private AdminUtilsOperation adminUtilsOperation;
+
+	private static final Log logger = LogFactory.getLog(KStreamBinderConfiguration.class);
+
+	@Bean
+	public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) {
+		return new KafkaTopicProvisioner(binderConfigurationProperties, adminUtilsOperation);
+	}
+
+	@Bean
+	public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties,
+			KafkaTopicProvisioner kafkaTopicProvisioner,
+			KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
+		return new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties,
+				streamsConfig);
+	}
+
+	@Bean(name = "adminUtilsOperation")
+	@Conditional(Kafka09Present.class)
+	@ConditionalOnClass(name = "kafka.admin.AdminUtils")
+	public AdminUtilsOperation kafka09AdminUtilsOperation() {
+		logger.info("AdminUtils selected: Kafka 0.9 AdminUtils");
+		return new Kafka09AdminUtilsOperation();
+	}
+
+	@Bean(name = "adminUtilsOperation")
+	@Conditional(Kafka10Present.class)
+	@ConditionalOnClass(name = "kafka.admin.AdminUtils")
+	public AdminUtilsOperation kafka10AdminUtilsOperation() {
+		logger.info("AdminUtils selected: Kafka 0.10 AdminUtils");
+		return new Kafka10AdminUtilsOperation();
+	}
+
+	static class Kafka10Present implements Condition {
+
+		@Override
+		public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
+			return AppInfoParser.getVersion().startsWith("0.10");
+		}
+	}
+
+	static class Kafka09Present implements Condition {
+
+		@Override
+		public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
+			return AppInfoParser.getVersion().startsWith("0.9");
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+import java.util.Properties;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.UnsatisfiedDependencyException;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.kstream.KStreamBoundElementFactory;
+import org.springframework.cloud.stream.binder.kstream.KStreamListenerParameterAdapter;
+import org.springframework.cloud.stream.binder.kstream.KStreamStreamListenerResultAdapter;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.codec.Codec;
+import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.core.KStreamBuilderFactoryBean;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamBinderSupportAutoConfiguration {
+
+	@Bean
+	@ConfigurationProperties(prefix = "spring.cloud.stream.kstream.binder")
+	public KafkaBinderConfigurationProperties binderConfigurationProperties() {
+		return new KafkaBinderConfigurationProperties();
+	}
+
+	@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME)
+	public KStreamBuilderFactoryBean defaultKStreamBuilder(
+			@Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider<StreamsConfig> streamsConfigProvider) {
+		StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable();
+		if (streamsConfig != null) {
+			KStreamBuilderFactoryBean kStreamBuilderFactoryBean = new KStreamBuilderFactoryBean(streamsConfig);
+			kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500);
+			return kStreamBuilderFactoryBean;
+		}
+		else {
+			throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(),
+					KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME, "streamsConfig",
+					"There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME
+							+ "' StreamsConfig bean in the application context.\n");
+		}
+	}
+
+	@Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+	public StreamsConfig streamsConfig(KafkaBinderConfigurationProperties binderConfigurationProperties) {
+		Properties props = new Properties();
+		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString());
+		props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
+		props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
+		props.put(StreamsConfig.APPLICATION_ID_CONFIG, "default");
+		props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, binderConfigurationProperties.getZkConnectionString());
+		if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) {
+			props.putAll(binderConfigurationProperties.getConfiguration());
+		}
+		return new StreamsConfig(props);
+	}
+
+	@Bean
+	public KStreamStreamListenerResultAdapter kStreamStreamListenerResultAdapter() {
+		return new KStreamStreamListenerResultAdapter();
+	}
+
+	@Bean
+	public KStreamListenerParameterAdapter kStreamListenerParameterAdapter(
+			CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		return new KStreamListenerParameterAdapter(
+				compositeMessageConverterFactory.getMessageConverterForAllRegistered());
+	}
+
+	@Bean
+	public KStreamBoundElementFactory kStreamBindableTargetFactory(KStreamBuilder kStreamBuilder,
+			BindingServiceProperties bindingServiceProperties, Codec codec,
+			CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties, codec,
+				compositeMessageConverterFactory);
+	}
+
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBindingProperties.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamBindingProperties {
+
+	private KStreamConsumerProperties consumer = new KStreamConsumerProperties();
+
+	private KStreamProducerProperties producer = new KStreamProducerProperties();
+
+	public KStreamConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(KStreamConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+
+	public KStreamProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(KStreamProducerProperties producer) {
+		this.producer = producer;
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+/**
+ * @author Soby Chacko
+ */
+public class KStreamCommonProperties {
+
+	private String keySerde;
+
+	private String valueSerde;
+
+	public String getKeySerde() {
+		return keySerde;
+	}
+
+	public void setKeySerde(String keySerde) {
+		this.keySerde = keySerde;
+	}
+
+	public String getValueSerde() {
+		return valueSerde;
+	}
+
+	public void setValueSerde(String valueSerde) {
+		this.valueSerde = valueSerde;
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamConsumerProperties extends KStreamCommonProperties {
+
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamExtendedBindingProperties.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
+
+/**
+ * @author Marius Bogoevici
+ */
+@ConfigurationProperties("spring.cloud.stream.kstream")
+public class KStreamExtendedBindingProperties
+		implements ExtendedBindingProperties<KStreamConsumerProperties, KStreamProducerProperties> {
+
+	private Map<String, KStreamBindingProperties> bindings = new HashMap<>();
+
+	public Map<String, KStreamBindingProperties> getBindings() {
+		return this.bindings;
+	}
+
+	public void setBindings(Map<String, KStreamBindingProperties> bindings) {
+		this.bindings = bindings;
+	}
+
+	@Override
+	public KStreamConsumerProperties getExtendedConsumerProperties(String binding) {
+		if (this.bindings.containsKey(binding) && this.bindings.get(binding).getConsumer() != null) {
+			return this.bindings.get(binding).getConsumer();
+		}
+		else {
+			return new KStreamConsumerProperties();
+		}
+	}
+
+	@Override
+	public KStreamProducerProperties getExtendedProducerProperties(String binding) {
+		if (this.bindings.containsKey(binding) && this.bindings.get(binding).getProducer() != null) {
+			return this.bindings.get(binding).getProducer();
+		}
+		else {
+			return new KStreamProducerProperties();
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2017 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.kstream.config;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KStreamProducerProperties extends KStreamCommonProperties {
+
+}

spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders

@@ -0,0 +1,4 @@
+kstream:\
+org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfiguration
+
+

spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,4 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+  org.springframework.cloud.stream.binder.kstream.config.KStreamBinderSupportAutoConfiguration
+
+  

spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.util.Map;
+
+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.common.serialization.LongDeserializer;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Predicate;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.Windowed;
+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.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.handler.annotation.SendTo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Soby Chacko
+ */
+public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id");
+
+	private static Consumer<Integer, Long> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<Integer, Long> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	@Test
+	public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception {
+		SpringApplication app = new SpringApplication(ProductCountApplication.class);
+		app.setWebEnvironment(false);
+		ConfigurableApplicationContext context = app.run("--server.port=0",
+				"--spring.cloud.stream.bindings.input.destination=foos",
+				"--spring.cloud.stream.bindings.output.destination=counts-id",
+				"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.bindings.output.producer.headerMode=raw",
+				"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
+				"--spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
+				"--spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
+				"--spring.cloud.stream.bindings.input.consumer.headerMode=raw",
+				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
+		receiveAndValidateFoo(context);
+		context.close();
+	}
+
+	private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic("foos");
+		template.sendDefault("{\"id\":\"123\"}");
+		ConsumerRecord<Integer, Long> cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id");
+
+		assertThat(cr.key().equals(123));
+		assertThat(cr.value().equals(1L));
+	}
+
+	@EnableBinding(KStreamProcessor.class)
+	@EnableAutoConfiguration
+	public static class ProductCountApplication {
+
+		@StreamListener("input")
+		@SendTo("output")
+		public KStream<Integer, Long> process(KStream<Object, Product> input) {
+			return input
+					.filter(new Predicate<Object, Product>() {
+
+						@Override
+						public boolean test(Object key, Product product) {
+							return product.getId() == 123;
+						}
+					})
+					.map(new KeyValueMapper<Object, Product, KeyValue<Product, Product>>() {
+
+						@Override
+						public KeyValue<Product, Product> apply(Object key, Product value) {
+							return new KeyValue<>(value, value);
+						}
+					})
+					.groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))
+					.count(TimeWindows.of(5000), "id-count-store")
+					.toStream()
+					.map(new KeyValueMapper<Windowed<Product>, Long, KeyValue<Integer, Long>>() {
+
+						@Override
+						public KeyValue<Integer, Long> apply(Windowed<Product> key, Long value) {
+							return new KeyValue<>(key.key().id, value);
+						}
+					});
+		}
+	}
+
+	static class Product {
+
+		Integer id;
+
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java

@@ -0,0 +1,253 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+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.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KStreamBuilderFactoryBean;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.handler.annotation.SendTo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Marius Bogoevici
+ * @author Soby Chacko
+ */
+public class KStreamBinderWordCountIntegrationTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts");
+
+	private static Consumer<String, String> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	@Test
+	public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception {
+		SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
+		app.setWebEnvironment(false);
+
+		ConfigurableApplicationContext context = app.run("--server.port=0",
+				"--spring.cloud.stream.bindings.input.destination=words",
+				"--spring.cloud.stream.bindings.output.destination=counts",
+				"--spring.cloud.stream.bindings.output.contentType=application/json",
+				"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.bindings.output.producer.headerMode=raw",
+				"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
+				"--spring.cloud.stream.bindings.input.consumer.headerMode=raw",
+				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
+		receiveAndValidate(context);
+		context.close();
+	}
+
+	private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception{
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic("words");
+		template.sendDefault("foobar");
+		ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts");
+		assertThat(cr.value().contains("\"word\":\"foobar\",\"count\":1")).isTrue();
+	}
+
+	@EnableBinding(KStreamProcessor.class)
+	@EnableAutoConfiguration
+	@EnableConfigurationProperties(WordCountProcessorProperties.class)
+	public static class WordCountProcessorApplication {
+
+		@Autowired
+		private WordCountProcessorProperties processorProperties;
+
+		@Autowired
+		private KStreamBuilderFactoryBean kafkaStreams;
+
+		@StreamListener("input")
+		@SendTo("output")
+		public KStream<?, WordCount> process(KStream<Object, String> input) {
+
+			return input
+					.flatMapValues(new ValueMapper<String, Iterable<String>>() {
+
+						@Override
+						public List<String> apply(String value) {
+							return Arrays.asList(value.toLowerCase().split("\\W+"));
+						}
+					})
+					.map(new KeyValueMapper<Object, String, KeyValue<String, String>>() {
+
+						@Override
+						public KeyValue<String, String> apply(Object key, String value) {
+							return new KeyValue<>(value, value);
+						}
+					})
+					.groupByKey(Serdes.String(), Serdes.String())
+					.count(configuredTimeWindow(), processorProperties.getStoreName())
+					.toStream()
+					.map(new KeyValueMapper<Windowed<String>, Long, KeyValue<Object, WordCount>>() {
+
+						@Override
+						public KeyValue<Object, WordCount> apply(Windowed<String> key, Long value) {
+							return new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())));
+						}
+					});
+		}
+
+		/**
+		 * Constructs a {@link TimeWindows} property.
+		 *
+		 * @return
+		 */
+		private TimeWindows configuredTimeWindow() {
+			return processorProperties.getAdvanceBy() > 0
+					? TimeWindows.of(processorProperties.getWindowLength()).advanceBy(processorProperties.getAdvanceBy())
+					: TimeWindows.of(processorProperties.getWindowLength());
+		}
+	}
+
+	@ConfigurationProperties(prefix = "kstream.word.count")
+	static class  WordCountProcessorProperties {
+
+		private int windowLength = 5000;
+
+		private int advanceBy = 0;
+
+		private String storeName = "WordCounts";
+
+		int getWindowLength() {
+			return windowLength;
+		}
+
+		public void setWindowLength(int windowLength) {
+			this.windowLength = windowLength;
+		}
+
+		int getAdvanceBy() {
+			return advanceBy;
+		}
+
+		public void setAdvanceBy(int advanceBy) {
+			this.advanceBy = advanceBy;
+		}
+
+		String getStoreName() {
+			return storeName;
+		}
+
+		public void setStoreName(String storeName) {
+			this.storeName = storeName;
+		}
+	}
+
+	static class WordCount {
+
+		private String word;
+
+		private long count;
+
+		private Date start;
+
+		private Date end;
+
+		WordCount(String word, long count, Date start, Date end) {
+			this.word = word;
+			this.count = count;
+			this.start = start;
+			this.end = end;
+		}
+
+		public String getWord() {
+			return word;
+		}
+
+		public void setWord(String word) {
+			this.word = word;
+		}
+
+		public long getCount() {
+			return count;
+		}
+
+		public void setCount(long count) {
+			this.count = count;
+		}
+
+		public Date getStart() {
+			return start;
+		}
+
+		public void setStart(Date start) {
+			this.start = start;
+		}
+
+		public Date getEnd() {
+			return end;
+		}
+
+		public void setEnd(Date end) {
+			this.end = end;
+		}
+	}
+
+}

spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.util.Map;
+
+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.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Predicate;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
+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.KStreamBuilderFactoryBean;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.handler.annotation.SendTo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+public class KStreamInteractiveQueryIntegrationTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id");
+
+	private static Consumer<String, String> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	@Test
+	public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception {
+		SpringApplication app = new SpringApplication(ProductCountApplication.class);
+		app.setWebEnvironment(false);
+		ConfigurableApplicationContext context = app.run("--server.port=0",
+				"--spring.cloud.stream.bindings.input.destination=foos",
+				"--spring.cloud.stream.bindings.output.destination=counts-id",
+				"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.bindings.output.producer.headerMode=raw",
+				"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
+				"--spring.cloud.stream.bindings.input.consumer.headerMode=raw",
+				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
+		receiveAndValidateFoo(context);
+		context.close();
+	}
+
+	private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic("foos");
+		template.sendDefault("{\"id\":\"123\"}");
+		ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id");
+		assertThat(cr.value().contains("Count for product with ID 123: 1")).isTrue();
+
+		ProductCountApplication.Foo foo = context.getBean(ProductCountApplication.Foo.class);
+		assertThat(foo.getProductStock(123).equals(1L));
+	}
+
+	@EnableBinding(KStreamProcessor.class)
+	@EnableAutoConfiguration
+	public static class ProductCountApplication {
+
+		@Autowired
+		private KStreamBuilderFactoryBean kStreamBuilderFactoryBean;
+
+		@StreamListener("input")
+		@SendTo("output")
+		public KStream<?, String> process(KStream<Object, Product> input) {
+
+			return input
+					.filter(new Predicate<Object, Product>() {
+
+						@Override
+						public boolean test(Object key, Product product) {
+							return product.getId() == 123;
+						}
+					})
+					.map(new KeyValueMapper<Object, Product, KeyValue<Integer, Product>>() {
+
+						@Override
+						public KeyValue<Integer, Product> apply(Object key, Product value) {
+							return new KeyValue<>(value.id, value);
+						}
+					})
+					.groupByKey(new Serdes.IntegerSerde(), new JsonSerde<>(Product.class))
+					.count("prod-id-count-store")
+					.toStream()
+					.map(new KeyValueMapper<Integer, Long, KeyValue<Object, String>>() {
+
+						@Override
+						public KeyValue<Object, String> apply(Integer key, Long value) {
+							return new KeyValue<>(null, "Count for product with ID 123: " + value);
+						}
+					});
+		}
+
+		@Bean
+		public Foo foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) {
+			return new Foo(kStreamBuilderFactoryBean);
+		}
+
+
+		static class Foo {
+			KStreamBuilderFactoryBean kStreamBuilderFactoryBean;
+
+			Foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) {
+				this.kStreamBuilderFactoryBean = kStreamBuilderFactoryBean;
+			}
+
+			public Long getProductStock(Integer id) {
+				KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams();
+				ReadOnlyKeyValueStore<Object, Object> keyValueStore =
+						streams.store("prod-id-count-store", QueryableStoreTypes.keyValueStore());
+				return (Long)keyValueStore.get(id);
+			}
+		}
+
+	}
+
+	static class Product {
+
+		Integer id;
+
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java

@@ -0,0 +1,153 @@
+/*
+ * Copyright 2017 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.kstream;
+
+import java.util.Map;
+
+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.streams.KeyValue;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Predicate;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.Windowed;
+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.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.handler.annotation.SendTo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Marius Bogoevici
+ * @author Soby Chacko
+ */
+public class KstreamBinderPojoInputStringOutputIntegrationTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id");
+
+	private static Consumer<String, String> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	@Test
+	public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception {
+		SpringApplication app = new SpringApplication(ProductCountApplication.class);
+		app.setWebEnvironment(false);
+		ConfigurableApplicationContext context = app.run("--server.port=0",
+				"--spring.cloud.stream.bindings.input.destination=foos",
+				"--spring.cloud.stream.bindings.output.destination=counts-id",
+				"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+				"--spring.cloud.stream.bindings.output.producer.headerMode=raw",
+				"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
+				"--spring.cloud.stream.bindings.input.consumer.headerMode=raw",
+				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
+		receiveAndValidateFoo(context);
+		context.close();
+	}
+
+	private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic("foos");
+		template.sendDefault("{\"id\":\"123\"}");
+		ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id");
+		assertThat(cr.value().contains("Count for product with ID 123: 1")).isTrue();
+	}
+
+	@EnableBinding(KStreamProcessor.class)
+	@EnableAutoConfiguration
+	public static class ProductCountApplication {
+
+		@StreamListener("input")
+		@SendTo("output")
+		public KStream<?, String> process(KStream<Object, Product> input) {
+
+			return input
+					.filter(new Predicate<Object, Product>() {
+
+						@Override
+						public boolean test(Object key, Product product) {
+							return product.getId() == 123;
+						}
+					})
+					.map(new KeyValueMapper<Object, Product, KeyValue<Product, Product>>() {
+
+						@Override
+						public KeyValue<Product, Product> apply(Object key, Product value) {
+							return new KeyValue<>(value, value);
+						}
+					})
+					.groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))
+					.count(TimeWindows.of(5000), "id-count-store")
+					.toStream()
+					.map(new KeyValueMapper<Windowed<Product>, Long, KeyValue<Object, String>>() {
+
+						@Override
+						public KeyValue<Object, String> apply(Windowed<Product> key, Long value) {
+							return new KeyValue<>(null, "Count for product with ID 123: " + value);
+						}
+					});
+		}
+	}
+
+	static class Product {
+
+		Integer id;
+
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+	}
+}

spring-cloud-stream-binder-kstream/src/test/resources/logback.xml

@@ -0,0 +1,15 @@
+<configuration>
+    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
+        </encoder>
+    </appender>
+    <logger name="org.springframework.integration.kafka" level="INFO"/>
+    <logger name="org.springframework.kafka" level="INFO"/>
+    <logger name="org.springframework.cloud.stream" level="INFO" />
+    <logger name="org.springframework.integration.channel" level="INFO" />
+    <root level="WARN">
+        <appender-ref ref="stdout"/>
+    </root>
+</configuration>
+