1.3.3.RELEASE

ConfigurationProperties bean provided by Kafka Streams binder
extends from `KafkaBinderConfigurationProperties` used by Kafka binder.
It creates a conflict when autowiring this bean from Kafka binder configuration.
This prevents an application to have both binders in the classpath.
Change the creation of this ConfigurationProperties bean so that it
avoids creating bean using EnbaleConfigurationProperties and then autowiring,
but directly create the Bean using `@Bean`. This prevents the conflict.

Resolves #244
Resolves #315

pom.xml

@@ -2,21 +2,21 @@
 <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.M2</version>
+	<version>1.3.3.RELEASE</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>1.3.3.RELEASE</version>
+		<version>1.3.9.RELEASE</version>
 		<relativePath />
 	</parent>
 	<properties>
 		<java.version>1.7</java.version>
 		<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.M2</spring-cloud-stream.version>
-		<spring-cloud-build.version>1.3.3.RELEASE</spring-cloud-build.version>
+		<spring-integration-kafka.version>2.1.2.RELEASE</spring-integration-kafka.version>
+		<spring-cloud-stream.version>1.3.3.RELEASE</spring-cloud-stream.version>
+		<spring-cloud-build.version>1.3.9.RELEASE</spring-cloud-build.version>
 	</properties>
 	<modules>
 		<module>spring-cloud-stream-binder-kafka</module>

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.M2</version>
+		<version>1.3.3.RELEASE</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.M2</version>
+        <version>1.3.3.RELEASE</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-0.10.1-test</artifactId>
     <description>Spring Cloud Stream Kafka Binder 0.10.1 Tests</description>
@@ -22,12 +22,10 @@
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
-            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka</artifactId>
-            <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -60,6 +58,14 @@
             <groupId>org.springframework.integration</groupId>
             <artifactId>spring-integration-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-jmx</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka</artifactId>

spring-cloud-stream-binder-kafka-0.10.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_1_BinderTests.java

@@ -16,13 +16,6 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication;
 import kafka.utils.ZKStringSerializer$;
@@ -36,7 +29,6 @@ import org.eclipse.jetty.server.Server;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
-
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -58,11 +50,18 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.MessageBuilder;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
 import static org.junit.Assert.assertTrue;
 
 /**
  * Integration tests for the {@link KafkaMessageChannelBinder}.
- * 
+ *
  * This test specifically tests for the 0.10.1.x version of Kafka.
  *
  * @author Eric Bottard
@@ -79,7 +78,7 @@ public class Kafka_0_10_1_BinderTests extends Kafka_0_10_2_BinderTests {
 
 	private Kafka10TestBinder binder;
 
-	private Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
+	private final Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
 
 	@Override
 	protected void binderBindUnbindLatency() throws InterruptedException {
@@ -90,11 +89,13 @@ public class Kafka_0_10_1_BinderTests extends Kafka_0_10_2_BinderTests {
 	protected Kafka10TestBinder getBinder() {
 		if (binder == null) {
 			KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
+			binderConfiguration.setHeaders("dlqTestHeader");
 			binder = new Kafka10TestBinder(binderConfiguration);
 		}
 		return binder;
 	}
 
+	@Override
 	protected KafkaBinderConfigurationProperties createConfigurationProperties() {
 		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
 		BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();

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.M2</version>
+        <version>1.3.3.RELEASE</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-0.10.2-test</artifactId>
     <description>Spring Cloud Stream Kafka Binder 0.10.2 Tests</description>
@@ -23,12 +23,10 @@
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
-            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka</artifactId>
-            <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -61,6 +59,14 @@
             <groupId>org.springframework.integration</groupId>
             <artifactId>spring-integration-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-jmx</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka</artifactId>

spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java

@@ -67,6 +67,7 @@ public class Kafka10TestBinder extends AbstractKafkaTestBinder {
 			ProducerListener producerListener = new LoggingProducerListener();
 			binder.setProducerListener(producerListener);
 			AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
+			setApplicationContext(context);
 			binder.setApplicationContext(context);
 			binder.afterPropertiesSet();
 			this.setBinder(binder);

spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_2_BinderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-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.
@@ -25,8 +25,10 @@ import java.util.UUID;
 
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication;
+
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
+
 import org.I0Itec.zkclient.ZkClient;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -69,6 +71,7 @@ import static org.junit.Assert.assertTrue;
  * @author Marius Bogoevici
  * @author Mark Fisher
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
 
@@ -79,7 +82,7 @@ public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
 
 	private Kafka10TestBinder binder;
 
-	private Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
+	private final Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
 
 	@Override
 	protected void binderBindUnbindLatency() throws InterruptedException {
@@ -90,11 +93,13 @@ public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
 	protected Kafka10TestBinder getBinder() {
 		if (binder == null) {
 			KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
+			binderConfiguration.setHeaders("dlqTestHeader");
 			binder = new Kafka10TestBinder(binderConfiguration);
 		}
 		return binder;
 	}
 
+	@Override
 	protected KafkaBinderConfigurationProperties createConfigurationProperties() {
 		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
 		BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();

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.M2</version>
+        <version>1.3.3.RELEASE</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
     <description>Spring Cloud Stream Kafka Binder Core</description>

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -35,6 +35,7 @@ import org.springframework.util.StringUtils;
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
  * @author Soby Chacko
+ * @author Gary Russell
  */
 @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
 public class KafkaBinderConfigurationProperties {
@@ -88,6 +89,11 @@ public class KafkaBinderConfigurationProperties {
 
 	private int queueSize = 8192;
 
+	/**
+	 * Time to wait to get partition information in seconds; default 60.
+	 */
+	private int healthTimeout = 60;
+
 	private JaasLoginModuleConfiguration jaas;
 
 	public String getZkConnectionString() {
@@ -228,6 +234,14 @@ public class KafkaBinderConfigurationProperties {
 		this.minPartitionCount = minPartitionCount;
 	}
 
+	public int getHealthTimeout() {
+		return this.healthTimeout;
+	}
+
+	public void setHealthTimeout(int healthTimeout) {
+		this.healthTimeout = healthTimeout;
+	}
+
 	public int getQueueSize() {
 		return this.queueSize;
 	}

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.M2</version>
+        <version>1.3.3.RELEASE</version>
     </parent>
 
     <artifactId>spring-cloud-stream-binder-kafka-docs</artifactId>

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

@@ -23,12 +23,13 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
 
 = Reference Guide
 include::overview.adoc[]
+
 include::dlq.adoc[]
-include::metrics.adoc[]
 
 = Appendices
 [appendix]
 include::building.adoc[]
+
 include::contributing.adoc[]
 
 // ======================================================================================

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

@@ -1,10 +0,0 @@
-[[kafka-metrics]]
-== Kafka metrics
-
-Kafka binder module exposes the following metrics:
-
-`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag`  - this metric indicates how many messages
-have not been yet consumed from given binder's topic (`someTopic`) by given consumer group (`someGroup`).
-For example if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, then
-consumer group `myGroup` has `1000` messages to waiting to be consumed from topic `myTopic`. This metric is
-particularly useful to provide auto-scaling feedback to PaaS platform of your choice.

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

@@ -73,6 +73,11 @@ spring.cloud.stream.kafka.binder.headers::
   The list of custom headers that will be transported by the binder.
 +
 Default: empty.
+spring.cloud.stream.kafka.binder.healthTimeout::
+  The time to wait to get partition information in seconds; default 60.
+  Health will report as down if this timer expires.
++
+Default: 10.
 spring.cloud.stream.kafka.binder.offsetUpdateTimeWindow::
   The frequency, in milliseconds, with which offsets are saved.
 Ignored if `0`.
@@ -523,5 +528,53 @@ spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.c
 spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde
 ----
 
+timewindow.length::
+  Many streaming applications written using Kafka Streams involve windowning operations.
+  If you specify this property, there is a `org.apache.kafka.streams.kstream.TimeWindows` bean automatically provided that can be autowired in applications.
+  This property must be prefixed with `spring.cloud.stream.kstream.`.
+  A bean of type `org.apache.kafka.streams.kstream.TimeWindows` is created only if this property is provided.
+
+ Following is an example of using this property.
+ Values are provided in milliseconds.
+
+[source]
+----
+spring.cloud.stream.kstream.timeWindow.length=5000
+----
+
+timewindow.advanceBy::
+  This property goes hand in hand with `timewindow.length` and has no effect on its own.
+  If you provide this property, the generated `org.apache.kafka.streams.kstream.TimeWindows` bean will automatically conatin this information.
+  This property must be prefixed with `spring.cloud.stream.kstream.`.
+
+ Following is an example of using this property.
+ Values are provided in milliseconds.
+
+[source]
+----
+spring.cloud.stream.kstream.timeWindow.advanceBy=1000
+----
 
 
+[[kafka-error-channels]]
+== Error Channels
+
+Starting with _version 1.3_, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too.
+See <<binder-error-channels>> for more information.
+
+The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureException` with properties:
+
+* `failedMessage` - the spring-messaging `Message<?>` that failed to be sent.
+* `record` - the raw `ProducerRecord` that was created from the `failedMessage`
+
+There is no automatic handling of these exceptions (such as sending to a <<kafka-dlq-processing, Dead-Letter queue>>); you can consume these exceptions with your own Spring Integration flow.
+
+
+[[kafka-metrics]]
+== Kafka Metrics
+
+Kafka binder module exposes the following metrics:
+
+`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag`  - this metric indicates how many messages have not been yet consumed from given binder's topic by given consumer group.
+For example if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, then consumer group `myGroup` has `1000` messages to waiting to be consumed from topic `myTopic`.
+This metric is particularly useful to provide auto-scaling feedback to PaaS platform of your choice.

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.M2</version>
+		<version>1.3.3.RELEASE</version>
 	</parent>
 
 	<dependencies>
@@ -52,11 +52,6 @@
 		<dependency>
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka</artifactId>
-			<version>${spring-kafka.version}</version>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework.integration</groupId>
-			<artifactId>spring-integration-kafka</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

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

@@ -19,6 +19,13 @@ package org.springframework.cloud.stream.binder.kafka;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.PartitionInfo;
@@ -33,13 +40,18 @@ import org.springframework.kafka.core.ConsumerFactory;
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
  * @author Henryk Konsek
+ * @author Gary Russell
  */
 public class KafkaBinderHealthIndicator implements HealthIndicator {
 
+	private static final int DEFAULT_TIMEOUT = 60;
+
 	private final KafkaMessageChannelBinder binder;
 
 	private final ConsumerFactory<?, ?> consumerFactory;
 
+	private int timeout = DEFAULT_TIMEOUT;
+
 	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
 			ConsumerFactory<?, ?> consumerFactory) {
 		this.binder = binder;
@@ -47,28 +59,67 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
 
 	}
 
+	/**
+	 * Set the timeout in seconds to retrieve health information.
+	 * @param timeout the timeout - default 60.
+	 */
+	public void setTimeout(int timeout) {
+		this.timeout = timeout;
+	}
+
 	@Override
 	public Health health() {
-		try (Consumer<?, ?> metadataConsumer = consumerFactory.createConsumer()) {
-			Set<String> downMessages = new HashSet<>();
-			for (String topic : this.binder.getTopicsInUse().keySet()) {
-				List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
-				for (PartitionInfo partitionInfo : partitionInfos) {
-					if (this.binder.getTopicsInUse().get(topic).getPartitionInfos().contains(partitionInfo)
-							&& partitionInfo.leader()
-									.id() == -1) {
-						downMessages.add(partitionInfo.toString());
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		Future<Health> future = exec.submit(new Callable<Health>() {
+
+			@Override
+			public Health call() {
+				try (Consumer<?, ?> metadataConsumer = consumerFactory.createConsumer()) {
+					Set<String> downMessages = new HashSet<>();
+					for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) {
+						List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
+						for (PartitionInfo partitionInfo : partitionInfos) {
+							if (KafkaBinderHealthIndicator.this.binder.getTopicsInUse().get(topic).getPartitionInfos()
+									.contains(partitionInfo) && partitionInfo.leader().id() == -1) {
+								downMessages.add(partitionInfo.toString());
+							}
+						}
+					}
+					if (downMessages.isEmpty()) {
+						return Health.up().build();
+					}
+					else {
+						return Health.down()
+							.withDetail("Following partitions in use have no leaders: ", downMessages.toString())
+							.build();
 					}
 				}
+				catch (Exception e) {
+					return Health.down(e).build();
+				}
 			}
-			if (downMessages.isEmpty()) {
-				return Health.up().build();
-			}
-			return Health.down().withDetail("Following partitions in use have no leaders: ", downMessages.toString())
+
+		});
+		try {
+			return future.get(this.timeout, TimeUnit.SECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return Health.down()
+					.withDetail("Interrupted while waiting for partition information in", this.timeout + " seconds")
 					.build();
 		}
-		catch (Exception e) {
+		catch (ExecutionException e) {
 			return Health.down(e).build();
 		}
+		catch (TimeoutException e) {
+			return Health.down()
+					.withDetail("Failed to retrieve partition information in", this.timeout + " seconds")
+					.build();
+		}
+		finally {
+			exec.shutdownNow();
+		}
 	}
+
 }

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

@@ -16,17 +16,22 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Callable;
 
+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.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -36,9 +41,12 @@ import org.apache.kafka.common.utils.Utils;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderHeaders;
+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.KafkaExtendedBindingProperties;
@@ -68,6 +76,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -87,12 +96,19 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * @author Soby Chacko
  * @author Henryk Konsek
  * @author Doug Saus
+ * @author Aldo Sinanaj
  */
 public class KafkaMessageChannelBinder extends
 		AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,
 			ExtendedProducerProperties<KafkaProducerProperties>, KafkaTopicProvisioner>
 		implements ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
 
+	public static final String X_ORIGINAL_TOPIC = "x-original-topic";
+
+	public static final String X_EXCEPTION_MESSAGE = "x-exception-message";
+
+	public static final String X_EXCEPTION_STACKTRACE = "x-exception-stacktrace";
+
 	private final KafkaBinderConfigurationProperties configurationProperties;
 
 	private final Map<String, TopicInformation> topicsInUse = new HashMap<>();
@@ -147,16 +163,23 @@ public class KafkaMessageChannelBinder extends
 
 	@Override
 	protected MessageHandler createProducerMessageHandler(final ProducerDestination destination,
-			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties, MessageChannel errorChannel)
+					throws Exception {
 		final DefaultKafkaProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
 		Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(
 				producerProperties.getPartitionCount(),
 				false,
 				new Callable<Collection<PartitionInfo>>() {
+
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {
-						return producerFB.createProducer().partitionsFor(destination.getName());
+						Producer<byte[], byte[]> producer = producerFB.createProducer();
+						List<PartitionInfo> partitionsFor = producer.partitionsFor(destination.getName());
+						producer.close();
+						producerFB.destroy();
+						return partitionsFor;
 					}
+
 				});
 		this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions));
 		if (producerProperties.getPartitionCount() < partitions.size()) {
@@ -165,14 +188,27 @@ public class KafkaMessageChannelBinder extends
 						+ producerProperties.getPartitionCount() + ", smaller than the actual partition count of "
 						+ partitions.size() + " of the topic. The larger number will be used instead.");
 			}
+			/*
+			 * This is dirty; it relies on the fact that we, and the partition
+			 * interceptor, share a hard reference to the producer properties instance.
+			 * But I don't see another way to fix it since the interceptor has already
+			 * been added to the channel, and we don't have access to the channel here; if
+			 * we did, we could inject the proper partition count there.
+			 * TODO: Consider this when doing the 2.0 binder restructuring.
+			 */
+			producerProperties.setPartitionCount(partitions.size());
 		}
 
 		KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFB);
 		if (this.producerListener != null) {
 			kafkaTemplate.setProducerListener(this.producerListener);
 		}
-		return new ProducerConfigurationMessageHandler(kafkaTemplate, destination.getName(), producerProperties,
-				producerFB);
+		ProducerConfigurationMessageHandler handler = new ProducerConfigurationMessageHandler(kafkaTemplate,
+				destination.getName(), producerProperties, producerFB);
+		if (errorChannel != null) {
+			handler.setSendFailureChannel(errorChannel);
+		}
+		return handler;
 	}
 
 	private DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
@@ -224,10 +260,15 @@ public class KafkaMessageChannelBinder extends
 		Collection<PartitionInfo> allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount,
 				extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
 				new Callable<Collection<PartitionInfo>>() {
+
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {
-						return consumerFactory.createConsumer().partitionsFor(destination.getName());
+						Consumer<?, ?> consumer = consumerFactory.createConsumer();
+						List<PartitionInfo> partitionsFor = consumer.partitionsFor(destination.getName());
+						consumer.close();
+						return partitionsFor;
 					}
+
 				});
 
 		Collection<PartitionInfo> listenedPartitions;
@@ -315,10 +356,36 @@ public class KafkaMessageChannelBinder extends
 				public void handleMessage(Message<?> message) throws MessagingException {
 					final ConsumerRecord<?, ?> record = message.getHeaders()
 							.get(KafkaMessageDrivenChannelAdapter.KAFKA_RAW_DATA, ConsumerRecord.class);
-					final byte[] key = record.key() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) record.key()))
+					final byte[] key = record.key() != null
+							? Utils.toArray(ByteBuffer.wrap((byte[]) record.key()))
 							: null;
-					final byte[] payload = record.value() != null
-							? Utils.toArray(ByteBuffer.wrap((byte[]) record.value())) : null;
+					final byte[] payload;
+					if (HeaderMode.embeddedHeaders == extendedConsumerProperties.getHeaderMode()
+							&& message.getPayload() instanceof Throwable) {
+						final Throwable throwable = (Throwable) message.getPayload();
+						final String failureMessage = throwable.getMessage();
+
+						try {
+							MessageValues messageValues = EmbeddedHeaderUtils
+									.extractHeaders(MessageBuilder.withPayload((byte[]) record.value()).build(),
+											false);
+							messageValues.put(X_ORIGINAL_TOPIC, record.topic());
+							messageValues.put(X_EXCEPTION_MESSAGE, failureMessage);
+							messageValues.put(X_EXCEPTION_STACKTRACE, getStackTraceAsString(throwable));
+
+							final String[] headersToEmbed = new ArrayList<>(messageValues.keySet()).toArray(
+									new String[messageValues.keySet().size()]);
+							payload = EmbeddedHeaderUtils.embedHeaders(messageValues,
+									EmbeddedHeaderUtils.headersToEmbed(headersToEmbed));
+						}
+						catch (Exception e) {
+							throw new RuntimeException(e);
+						}
+					}
+					else {
+						payload = record.value() != null
+								? Utils.toArray(ByteBuffer.wrap((byte[]) record.value())) : null;
+					}
 					String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName())
 							? extendedConsumerProperties.getExtension().getDlqName()
 							: "error." + destination.getName() + "." + group;
@@ -406,6 +473,13 @@ public class KafkaMessageChannelBinder extends
 		return original.substring(0, maxCharacters) + "...";
 	}
 
+	private String getStackTraceAsString(Throwable cause) {
+		StringWriter stringWriter = new StringWriter();
+		PrintWriter printWriter = new PrintWriter(stringWriter, true);
+		cause.printStackTrace(printWriter);
+		return stringWriter.getBuffer().toString();
+	}
+
 	private final class ProducerConfigurationMessageHandler extends KafkaProducerMessageHandler<byte[], byte[]>
 			implements Lifecycle {
 
@@ -413,7 +487,7 @@ public class KafkaMessageChannelBinder extends
 
 		private final DefaultKafkaProducerFactory<byte[], byte[]> producerFactory;
 
-		private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
+		ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
 				ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
 				DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
 			super(kafkaTemplate);
@@ -422,7 +496,7 @@ public class KafkaMessageChannelBinder extends
 			setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
 			if (producerProperties.isPartitioned()) {
 				SpelExpressionParser parser = new SpelExpressionParser();
-				setPartitionIdExpression(parser.parseExpression("headers." + BinderHeaders.PARTITION_HEADER));
+				setPartitionIdExpression(parser.parseExpression("headers['" + BinderHeaders.PARTITION_HEADER + "']"));
 			}
 			if (producerProperties.getExtension().isSync()) {
 				setSync(true);

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

@@ -68,11 +68,12 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Ilayaperumal Gopinathan
  * @author Henryk Konsek
+ * @author Gary Russell
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
 @Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
-@EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class })
+@EnableConfigurationProperties({ KafkaExtendedBindingProperties.class })
 public class KafkaBinderConfiguration {
 
 	protected static final Log logger = LogFactory.getLog(KafkaBinderConfiguration.class);
@@ -96,14 +97,20 @@ public class KafkaBinderConfiguration {
 	private AdminUtilsOperation adminUtilsOperation;
 
 	@Bean
-	KafkaTopicProvisioner provisioningProvider() {
-		return new KafkaTopicProvisioner(this.configurationProperties, this.adminUtilsOperation);
+	KafkaBinderConfigurationProperties configurationProperties() {
+		return new KafkaBinderConfigurationProperties();
 	}
 
 	@Bean
-	KafkaMessageChannelBinder kafkaMessageChannelBinder() {
+	KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties configurationProperties) {
+		return new KafkaTopicProvisioner(configurationProperties, this.adminUtilsOperation);
+	}
+
+	@Bean
+	KafkaMessageChannelBinder kafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
+														KafkaTopicProvisioner provisioningProvider) {
 		KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
-				this.configurationProperties, provisioningProvider());
+				configurationProperties, provisioningProvider);
 		kafkaMessageChannelBinder.setCodec(this.codec);
 		kafkaMessageChannelBinder.setProducerListener(producerListener);
 		kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties);
@@ -117,7 +124,8 @@ public class KafkaBinderConfiguration {
 	}
 
 	@Bean
-	KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
+	KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder,
+											KafkaBinderConfigurationProperties configurationProperties) {
 		Map<String, Object> props = new HashMap<>();
 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
@@ -125,14 +133,18 @@ public class KafkaBinderConfiguration {
 			props.putAll(configurationProperties.getConsumerConfiguration());
 		}
 		if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
-			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
+			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());
 		}
 		ConsumerFactory<?, ?> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
-		return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, consumerFactory);
+		KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator(kafkaMessageChannelBinder,
+				consumerFactory);
+		indicator.setTimeout(configurationProperties.getHealthTimeout());
+		return indicator;
 	}
 
 	@Bean
-	public PublicMetrics kafkaBinderMetrics(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
+	public PublicMetrics kafkaBinderMetrics(KafkaMessageChannelBinder kafkaMessageChannelBinder,
+										KafkaBinderConfigurationProperties configurationProperties) {
 		return new KafkaBinderMetrics(kafkaMessageChannelBinder, configurationProperties);
 	}
 

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java

@@ -15,24 +15,16 @@
  */
 package org.springframework.cloud.stream.binder.kafka;
 
-import java.util.List;
-
 import org.springframework.cloud.stream.binder.AbstractTestBinder;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
-import org.springframework.context.support.GenericApplicationContext;
-import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.context.ApplicationContext;
 import org.springframework.integration.codec.Codec;
-import org.springframework.integration.codec.kryo.KryoRegistrar;
 import org.springframework.integration.codec.kryo.PojoCodec;
-import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.tuple.TupleKryoRegistrar;
 
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.Registration;
-
 /**
  * @author Soby Chacko
  * @author Gary Russell
@@ -40,33 +32,23 @@ import com.esotericsoftware.kryo.Registration;
 public abstract class AbstractKafkaTestBinder extends
 		AbstractTestBinder<KafkaMessageChannelBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
 
+	private ApplicationContext applicationContext;
+
 	@Override
 	public void cleanup() {
 		// do nothing - the rule will take care of that
 	}
 
-	protected void addErrorChannel(GenericApplicationContext context) {
-		PublishSubscribeChannel errorChannel = new PublishSubscribeChannel();
-		context.getBeanFactory().initializeBean(errorChannel, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
-		context.getBeanFactory().registerSingleton(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, errorChannel);
+	protected final void setApplicationContext(ApplicationContext context) {
+		this.applicationContext = context;
+	}
+
+	public ApplicationContext getApplicationContext() {
+		return this.applicationContext;
 	}
 
 	protected static Codec getCodec() {
-		return new PojoCodec(new TupleRegistrar());
-	}
-
-	private static class TupleRegistrar implements KryoRegistrar {
-		private final TupleKryoRegistrar delegate = new TupleKryoRegistrar();
-
-		@Override
-		public void registerTypes(Kryo kryo) {
-			this.delegate.registerTypes(kryo);
-		}
-
-		@Override
-		public List<Registration> getRegistrations() {
-			return this.delegate.getRegistrations();
-		}
+		return new PojoCodec(new TupleKryoRegistrar());
 	}
 
 }

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java

@@ -15,6 +15,10 @@
  */
 package org.springframework.cloud.stream.binder.kafka;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.verify;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -27,16 +31,16 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.BDDMockito.given;
-
 /**
  * @author Barry Commins
+ * @author Gary Russell
  */
 public class KafkaBinderHealthIndicatorTest {
 
@@ -53,14 +57,15 @@ public class KafkaBinderHealthIndicatorTest {
 	@Mock
 	private KafkaMessageChannelBinder binder;
 
-	private Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = new HashMap<>();
+	private final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = new HashMap<>();
 
 	@Before
 	public void setup() {
 		MockitoAnnotations.initMocks(this);
 		given(consumerFactory.createConsumer()).willReturn(consumer);
 		given(binder.getTopicsInUse()).willReturn(topicsInUse);
-		indicator = new KafkaBinderHealthIndicator(binder, consumerFactory);
+		this.indicator = new KafkaBinderHealthIndicator(binder, consumerFactory);
+		this.indicator.setTimeout(10);
 	}
 
 	@Test
@@ -70,6 +75,7 @@ public class KafkaBinderHealthIndicatorTest {
 		given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
 		Health health = indicator.health();
 		assertThat(health.getStatus()).isEqualTo(Status.UP);
+		verify(this.consumer).close();
 	}
 
 	@Test
@@ -81,6 +87,25 @@ public class KafkaBinderHealthIndicatorTest {
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
 	}
 
+	@Test(timeout = 5000)
+	public void kafkaBinderDoesNotAnswer() {
+		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
+		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group", partitions));
+		given(consumer.partitionsFor(TEST_TOPIC)).willAnswer(new Answer<Object>() {
+
+			@Override
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				final int fiveMinutes = 1000 * 60 * 5;
+				Thread.sleep(fiveMinutes);
+				return partitions;
+			}
+
+		});
+		this.indicator.setTimeout(1);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
+	}
+
 	private List<PartitionInfo> partitions(Node leader) {
 		List<PartitionInfo> partitions = new ArrayList<>();
 		partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null));

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -16,6 +16,12 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -58,27 +64,32 @@ import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+import org.springframework.integration.kafka.support.KafkaSendFailureException;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.retry.backoff.FixedBackOffPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
-import static org.junit.Assert.assertTrue;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
@@ -87,6 +98,8 @@ import kafka.utils.ZkUtils;
  * @author Soby Chacko
  * @author Ilayaperumal Gopinathan
  * @author Henryk Konsek
+ * @author Gary Russell
+ * @author Aldo Sinanaj
  */
 public abstract class KafkaBinderTests extends
 		PartitionCapableBinderTests<AbstractKafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
@@ -130,16 +143,20 @@ public abstract class KafkaBinderTests extends
 
 	@Test
 	public void testDlqAndRetry() throws Exception {
-		testDlqGuts(true);
+		testDlqGuts(true, false);
 	}
 
 	@Test
 	public void testDlq() throws Exception {
-		testDlqGuts(false);
+		testDlqGuts(false, false);
 	}
 
-	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testDlqGuts(boolean withRetry) throws Exception {
+	@Test
+	public void testDlqRawHeaders() throws Exception {
+		testDlqGuts(false, true);
+	}
+
+	private void testDlqGuts(boolean withRetry, boolean raw) throws Exception {
 		AbstractKafkaTestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		DirectChannel moduleInputChannel = new DirectChannel();
@@ -148,12 +165,18 @@ public abstract class KafkaBinderTests extends
 		moduleInputChannel.subscribe(handler);
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setPartitionCount(2);
+		if (raw) {
+			producerProperties.setHeaderMode(HeaderMode.raw);
+		}
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setMaxAttempts(withRetry ? 2 : 1);
 		consumerProperties.setBackOffInitialInterval(100);
 		consumerProperties.setBackOffMaxInterval(150);
 		consumerProperties.getExtension().setEnableDlq(true);
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+		if (raw) {
+			consumerProperties.setHeaderMode(HeaderMode.raw);
+		}
 		long uniqueBindingId = System.currentTimeMillis();
 
 		String producerName = "dlqTest." + uniqueBindingId + ".0";
@@ -164,10 +187,12 @@ public abstract class KafkaBinderTests extends
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
 		dlqConsumerProperties.setMaxAttempts(1);
+		if (raw) {
+			dlqConsumerProperties.setHeaderMode(HeaderMode.raw);
+		}
 
 		ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(), "applicationContext",
 				ApplicationContext.class);
-		Map<String, MessageChannel> beansOfType = context.getBeansOfType(MessageChannel.class);
 		SubscribableChannel boundErrorChannel = context
 				.getBean(producerName + ".testGroup.errors-0", SubscribableChannel.class);
 		SubscribableChannel globalErrorChannel = context.getBean("errorChannel", SubscribableChannel.class);
@@ -197,13 +222,28 @@ public abstract class KafkaBinderTests extends
 				"error.dlqTest." + uniqueBindingId + ".0.testGroup", null, dlqChannel, dlqConsumerProperties);
 		binderBindUnbindLatency();
 		String testMessagePayload = "test." + UUID.randomUUID().toString();
-		Message<String> testMessage = MessageBuilder.withPayload(testMessagePayload).build();
+		Message<String> testMessage = MessageBuilder.withPayload(testMessagePayload)
+				.setHeader("dlqTestHeader", "propagatedToDlq")
+				.build();
 		moduleOutputChannel.send(testMessage);
 
 		Message<?> receivedMessage = receive(dlqChannel, 3);
 		assertThat(receivedMessage).isNotNull();
-		assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
-		assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
+		if (raw) {
+			assertThat(new String((byte[]) receivedMessage.getPayload(), StandardCharsets.UTF_8))
+					.isEqualTo(testMessagePayload);
+		}
+		else {
+			assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
+			final MessageHeaders headers = receivedMessage.getHeaders();
+			assertThat(headers.get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC)).isEqualTo(producerName);
+			assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_MESSAGE))
+					.isEqualTo("failed to send Message to channel 'null'; nested exception is java.lang.RuntimeException: fail");
+			assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_STACKTRACE)).isNotNull();
+			assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isNotNull();
+			assertThat(headers.get("dlqTestHeader")).isEqualTo("propagatedToDlq");
+			assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
+		}
 		binderBindUnbindLatency();
 
 		// verify we got a message on the dedicated error channel and the global (via bridge)
@@ -967,13 +1007,23 @@ public abstract class KafkaBinderTests extends
 
 	@Test
 	@Override
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public void testPartitionedModuleJava() throws Exception {
 		Binder binder = getBinder();
 
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+
+		final ZkClient zkClient;
+		zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
+				configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+
+		final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+		invokeCreateTopic(zkUtils, "partJ.0", 8, 1, new Properties());
+
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setConcurrency(2);
-		consumerProperties.setInstanceCount(3);
+		consumerProperties.setInstanceCount(4);
 		consumerProperties.setInstanceIndex(0);
 		consumerProperties.setPartitioned(true);
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
@@ -988,11 +1038,15 @@ public abstract class KafkaBinderTests extends
 		QueueChannel input2 = new QueueChannel();
 		input2.setBeanName("test.input2J");
 		Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties);
+		consumerProperties.setInstanceIndex(3);
+		QueueChannel input3 = new QueueChannel();
+		input3.setBeanName("test.input3J");
+		Binding<MessageChannel> input3Binding = binder.bindConsumer("partJ.0", "test", input3, consumerProperties);
 
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
 		producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
-		producerProperties.setPartitionCount(3);
+		producerProperties.setPartitionCount(3); // overridden to 8 on the actual topic
 		DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties));
 		output.setBeanName("test.output");
 		Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, producerProperties);
@@ -1005,6 +1059,7 @@ public abstract class KafkaBinderTests extends
 		output.send(new GenericMessage<>(2));
 		output.send(new GenericMessage<>(1));
 		output.send(new GenericMessage<>(0));
+		output.send(new GenericMessage<>(3));
 
 		Message<?> receive0 = receive(input0);
 		assertThat(receive0).isNotNull();
@@ -1012,20 +1067,18 @@ public abstract class KafkaBinderTests extends
 		assertThat(receive1).isNotNull();
 		Message<?> receive2 = receive(input2);
 		assertThat(receive2).isNotNull();
+		Message<?> receive3 = receive(input3);
+		assertThat(receive3).isNotNull();
 
-		if (usesExplicitRouting()) {
-			assertThat(receive0.getPayload()).isEqualTo(0);
-			assertThat(receive1.getPayload()).isEqualTo(1);
-			assertThat(receive2.getPayload()).isEqualTo(2);
-		}
-		else {
-			List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1, receive2);
-			assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2);
-		}
+		assertThat(receive0.getPayload()).isEqualTo(0);
+		assertThat(receive1.getPayload()).isEqualTo(1);
+		assertThat(receive2.getPayload()).isEqualTo(2);
+		assertThat(receive3.getPayload()).isEqualTo(3);
 
 		input0Binding.unbind();
 		input1Binding.unbind();
 		input2Binding.unbind();
+		input3Binding.unbind();
 		outputBinding.unbind();
 	}
 
@@ -1722,6 +1775,68 @@ public abstract class KafkaBinderTests extends
 		assertThat(extractEndpoint(producerBinding).isRunning()).isFalse();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testProducerErrorChannel() throws Exception {
+		AbstractKafkaTestBinder binder = getBinder();
+		DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties());
+		ExtendedProducerProperties<KafkaProducerProperties> producerProps = new ExtendedProducerProperties<>(
+				new KafkaProducerProperties());
+		producerProps.setHeaderMode(HeaderMode.raw);
+		producerProps.setErrorChannelEnabled(true);
+		Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps);
+		final Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
+				.build();
+		SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class);
+		final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
+		final CountDownLatch latch = new CountDownLatch(2);
+		ec.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				errorMessage.set(message);
+				latch.countDown();
+			}
+
+		});
+		SubscribableChannel globalEc = binder.getApplicationContext()
+				.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class);
+		globalEc.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				latch.countDown();
+			}
+
+		});
+		KafkaProducerMessageHandler endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle",
+				KafkaProducerMessageHandler.class);
+		final RuntimeException fooException = new RuntimeException("foo");
+		final AtomicReference<Object> sent = new AtomicReference<>();
+		new DirectFieldAccessor(endpoint).setPropertyValue("kafkaTemplate",
+				new KafkaTemplate(mock(ProducerFactory.class)) {
+
+					@Override
+					public ListenableFuture<SendResult> send(String topic, Object payload) {
+						sent.set(payload);
+						SettableListenableFuture<SendResult> future = new SettableListenableFuture<>();
+						future.setException(fooException);
+						return future;
+					}
+
+		});
+
+		moduleOutputChannel.send(message);
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
+		assertThat(errorMessage.get().getPayload()).isInstanceOf(KafkaSendFailureException.class);
+		KafkaSendFailureException exception = (KafkaSendFailureException) errorMessage.get().getPayload();
+		assertThat(exception.getCause()).isSameAs(fooException);
+		assertThat(new String(((byte[] )exception.getFailedMessage().getPayload()))).isEqualTo(message.getPayload());
+		assertThat(exception.getRecord().value()).isSameAs(sent.get());
+		producerBinding.unbind();
+	}
+
 	@Override
 	protected void binderBindUnbindLatency() throws InterruptedException {
 		Thread.sleep(500);

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

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

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

@@ -62,8 +62,11 @@ public class KStreamBinder extends
 
 	private final StreamsConfig streamsConfig;
 
+	private final KafkaBinderConfigurationProperties binderConfigurationProperties;
+
 	public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner,
-			KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
+						KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) {
+		this.binderConfigurationProperties = binderConfigurationProperties;
 		this.headers = EmbeddedHeaderUtils.headersToEmbed(binderConfigurationProperties.getHeaders());
 		this.kafkaTopicProvisioner = kafkaTopicProvisioner;
 		this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
@@ -72,7 +75,7 @@ public class KStreamBinder extends
 
 	@Override
 	protected Binding<KStream<Object, Object>> doBindConsumer(String name, String group,
-			KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
+															KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<KafkaConsumerProperties>(
 				new KafkaConsumerProperties());
@@ -83,17 +86,17 @@ public class KStreamBinder extends
 	@Override
 	@SuppressWarnings("unchecked")
 	protected Binding<KStream<Object, Object>> doBindProducer(String name, KStream<Object, Object> outboundBindTarget,
-			ExtendedProducerProperties<KStreamProducerProperties> properties) {
+															ExtendedProducerProperties<KStreamProducerProperties> properties) {
 		ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<KafkaProducerProperties>(
 				new KafkaProducerProperties());
-		this.kafkaTopicProvisioner.provisionProducerDestination(name , extendedProducerProperties);
+		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));
+							return new KeyValue<>(k, (Object) KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
 						}
 						catch (Exception e) {
 							throw new IllegalArgumentException(e);
@@ -111,7 +114,7 @@ public class KStreamBinder extends
 						.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));
+								return KeyValue.pair(k, (Object) KStreamBinder.this.serializePayloadIfNecessary((Message<?>) v));
 							}
 						});
 			}
@@ -126,31 +129,37 @@ public class KStreamBinder extends
 			}
 		}
 		if (!properties.isUseNativeEncoding() || StringUtils.hasText(properties.getExtension().getKeySerde()) || StringUtils.hasText(properties.getExtension().getValueSerde())) {
-			Serde<?> keySerde = Serdes.ByteArray();
-			Serde<?> valueSerde = Serdes.ByteArray();
 			try {
+				Serde<?> keySerde;
+				Serde<?> valueSerde;
+
 				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 {
+				else {
+					keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
+							Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
+				}
+
 				if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
 					valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
 					if (valueSerde instanceof Configurable) {
 						((Configurable) valueSerde).configure(streamsConfig.originals());
 					}
 				}
+				else {
+					valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
+							Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
+				}
+				outboundBindTarget.to((Serde<Object>) keySerde, (Serde<Object>) valueSerde, name);
 			}
 			catch (ClassNotFoundException e) {
 				throw new IllegalStateException("Serde class not found: ", e);
 			}
-			outboundBindTarget.to((Serde<Object>) keySerde, (Serde<Object>) valueSerde, name);
+
 		}
 		else {
 			outboundBindTarget.to(name);

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

@@ -0,0 +1,40 @@
+/*
+ * 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.kafka.streams.kstream.TimeWindows;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Soby Chacko
+ */
+@Configuration
+@EnableConfigurationProperties(KStreamApplicationSupportProperties.class)
+public class KStreamApplicationSupportAutoConfiguration {
+
+	@Bean
+	@ConditionalOnProperty("spring.cloud.stream.kstream.timeWindow.length")
+	public TimeWindows configuredTimeWindow(KStreamApplicationSupportProperties processorProperties) {
+		return processorProperties.getTimeWindow().getAdvanceBy() > 0
+				? TimeWindows.of(processorProperties.getTimeWindow().getLength()).advanceBy(processorProperties.getTimeWindow().getAdvanceBy())
+				: TimeWindows.of(processorProperties.getTimeWindow().getLength());
+	}
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * {@link ConfigurationProperties} that can be used by end user Kafka Stream applications. This class provides
+ * convenient ways to access the commonly used kafka stream properties from the user application. For example, windowing
+ * operations are common use cases in stream processing and one can provide window specific properties at runtime and use
+ * those properties in the applications using this class.
+ *
+ * @author Soby Chacko
+ */
+@ConfigurationProperties("spring.cloud.stream.kstream")
+public class KStreamApplicationSupportProperties {
+
+	private TimeWindow timeWindow;
+
+	public TimeWindow getTimeWindow() {
+		return timeWindow;
+	}
+
+	public void setTimeWindow(TimeWindow timeWindow) {
+		this.timeWindow = timeWindow;
+	}
+
+	public static class TimeWindow {
+
+		private int length;
+
+		private int advanceBy;
+
+		public int getLength() {
+			return length;
+		}
+
+		public void setLength(int length) {
+			this.length = length;
+		}
+
+		public int getAdvanceBy() {
+			return advanceBy;
+		}
+
+		public void setAdvanceBy(int advanceBy) {
+			this.advanceBy = advanceBy;
+		}
+	}
+}

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

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

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

@@ -39,15 +39,12 @@ 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;
@@ -96,6 +93,8 @@ public class KStreamBinderWordCountIntegrationTests {
 				"--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.timeWindow.length=5000",
+				"--spring.cloud.stream.kstream.timeWindow.advanceBy=0",
 				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
 				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
 		receiveAndValidate(context);
@@ -114,14 +113,10 @@ public class KStreamBinderWordCountIntegrationTests {
 
 	@EnableBinding(KStreamProcessor.class)
 	@EnableAutoConfiguration
-	@EnableConfigurationProperties(WordCountProcessorProperties.class)
 	public static class WordCountProcessorApplication {
 
 		@Autowired
-		private WordCountProcessorProperties processorProperties;
-
-		@Autowired
-		private KStreamBuilderFactoryBean kafkaStreams;
+		private TimeWindows timeWindows;
 
 		@StreamListener("input")
 		@SendTo("output")
@@ -143,7 +138,7 @@ public class KStreamBinderWordCountIntegrationTests {
 						}
 					})
 					.groupByKey(Serdes.String(), Serdes.String())
-					.count(configuredTimeWindow(), processorProperties.getStoreName())
+					.count(timeWindows, "WordCounts")
 					.toStream()
 					.map(new KeyValueMapper<Windowed<String>, Long, KeyValue<Object, WordCount>>() {
 
@@ -154,50 +149,6 @@ public class KStreamBinderWordCountIntegrationTests {
 					});
 		}
 
-		/**
-		 * 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 {

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

@@ -85,6 +85,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
 				"--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.bindings.input.consumer.headerMode=raw",
 				"--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(),
 				"--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
@@ -108,7 +109,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
 
 		@StreamListener("input")
 		@SendTo("output")
-		public KStream<?, String> process(KStream<Object, Product> input) {
+		public KStream<Integer, String> process(KStream<Object, Product> input) {
 
 			return input
 					.filter(new Predicate<Object, Product>() {
@@ -128,11 +129,11 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
 					.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>>() {
+					.map(new KeyValueMapper<Windowed<Product>, Long, KeyValue<Integer, String>>() {
 
 						@Override
-						public KeyValue<Object, String> apply(Windowed<Product> key, Long value) {
-							return new KeyValue<>(null, "Count for product with ID 123: " + value);
+						public KeyValue<Integer, String> apply(Windowed<Product> key, Long value) {
+							return new KeyValue<>(key.key().id, "Count for product with ID 123: " + value);
 						}
 					});
 		}