Merge pull request #612 from spring-operator/polish-urls-remaining-1.2.x

URL Cleanup

.settings.xml

@@ -21,7 +21,7 @@
 				<repository>
 					<id>spring-snapshots</id>
 					<name>Spring Snapshots</name>
-					<url>http://repo.spring.io/libs-snapshot-local</url>
+					<url>https://repo.spring.io/libs-snapshot-local</url>
 					<snapshots>
 						<enabled>true</enabled>
 					</snapshots>
@@ -29,7 +29,7 @@
 				<repository>
 					<id>spring-milestones</id>
 					<name>Spring Milestones</name>
-					<url>http://repo.spring.io/libs-milestone-local</url>
+					<url>https://repo.spring.io/libs-milestone-local</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>
@@ -37,7 +37,7 @@
 				<repository>
 					<id>spring-releases</id>
 					<name>Spring Releases</name>
-					<url>http://repo.spring.io/release</url>
+					<url>https://repo.spring.io/release</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>
@@ -47,7 +47,7 @@
 				<pluginRepository>
 					<id>spring-snapshots</id>
 					<name>Spring Snapshots</name>
-					<url>http://repo.spring.io/libs-snapshot-local</url>
+					<url>https://repo.spring.io/libs-snapshot-local</url>
 					<snapshots>
 						<enabled>true</enabled>
 					</snapshots>
@@ -55,7 +55,7 @@
 				<pluginRepository>
 					<id>spring-milestones</id>
 					<name>Spring Milestones</name>
-					<url>http://repo.spring.io/libs-milestone-local</url>
+					<url>https://repo.spring.io/libs-milestone-local</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>

LICENSE

@@ -1,6 +1,6 @@
                                  Apache License
                            Version 2.0, January 2004
-                        http://www.apache.org/licenses/
+                        https://www.apache.org/licenses/
 
    TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
 
@@ -192,7 +192,7 @@
    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
+       https://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,

mvnw

@@ -8,7 +8,7 @@
 # "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
+#    https://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

mvnw.cmd

@@ -7,7 +7,7 @@
 @REM "License"); you may not use this file except in compliance
 @REM with the License.  You may obtain a copy of the License at
 @REM
-@REM    http://www.apache.org/licenses/LICENSE-2.0
+@REM    https://www.apache.org/licenses/LICENSE-2.0
 @REM
 @REM Unless required by applicable law or agreed to in writing,
 @REM software distributed under the License is distributed on an

pom.xml

@@ -1,13 +1,13 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-	<version>1.2.0.M2</version>
+	<version>1.2.2.BUILD-SNAPSHOT</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>1.3.1.M1</version>
+		<version>1.3.3.BUILD-SNAPSHOT</version>
 		<relativePath />
 	</parent>
 	<properties>
@@ -15,8 +15,7 @@
 		<kafka.version>0.10.1.1</kafka.version>
 		<spring-kafka.version>1.1.2.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>2.1.0.RELEASE</spring-integration-kafka.version>
-		<spring-cloud-stream.version>1.2.0.M2</spring-cloud-stream.version>
-		<spring-boot.version>1.5.1.RELEASE</spring-boot.version>
+		<spring-cloud-stream.version>1.2.3.BUILD-SNAPSHOT</spring-cloud-stream.version>
 	</properties>
 	<modules>
 		<module>spring-cloud-stream-binder-kafka</module>
@@ -45,6 +44,10 @@
 				<artifactId>kafka_2.11</artifactId>
 				<version>${kafka.version}</version>
 				<exclusions>
+					<exclusion>
+						<groupId>jline</groupId>
+						<artifactId>jline</artifactId>
+					</exclusion>
 					<exclusion>
 						<groupId>org.slf4j</groupId>
 						<artifactId>slf4j-log4j12</artifactId>
@@ -135,7 +138,7 @@
 					<dependency>
 						<groupId>org.springframework.cloud</groupId>
 						<artifactId>spring-cloud-build-tools</artifactId>
-						<version>1.3.1.M1</version>
+						<version>1.3.3.BUILD-SNAPSHOT</version>
 					</dependency>
 				</dependencies>
 				<executions>
@@ -165,7 +168,7 @@
 				<repository>
 					<id>spring-snapshots</id>
 					<name>Spring Snapshots</name>
-					<url>http://repo.spring.io/libs-snapshot-local</url>
+					<url>https://repo.spring.io/libs-snapshot-local</url>
 					<snapshots>
 						<enabled>true</enabled>
 					</snapshots>
@@ -176,7 +179,7 @@
 				<repository>
 					<id>spring-milestones</id>
 					<name>Spring Milestones</name>
-					<url>http://repo.spring.io/libs-milestone-local</url>
+					<url>https://repo.spring.io/libs-milestone-local</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>
@@ -184,7 +187,7 @@
 				<repository>
 					<id>spring-releases</id>
 					<name>Spring Releases</name>
-					<url>http://repo.spring.io/release</url>
+					<url>https://repo.spring.io/release</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>
@@ -194,7 +197,7 @@
 				<pluginRepository>
 					<id>spring-snapshots</id>
 					<name>Spring Snapshots</name>
-					<url>http://repo.spring.io/libs-snapshot-local</url>
+					<url>https://repo.spring.io/libs-snapshot-local</url>
 					<snapshots>
 						<enabled>true</enabled>
 					</snapshots>
@@ -205,7 +208,7 @@
 				<pluginRepository>
 					<id>spring-milestones</id>
 					<name>Spring Milestones</name>
-					<url>http://repo.spring.io/libs-milestone-local</url>
+					<url>https://repo.spring.io/libs-milestone-local</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>
@@ -213,7 +216,7 @@
 				<pluginRepository>
 					<id>spring-releases</id>
 					<name>Spring Releases</name>
-					<url>http://repo.spring.io/libs-release-local</url>
+					<url>https://repo.spring.io/libs-release-local</url>
 					<snapshots>
 						<enabled>false</enabled>
 					</snapshots>

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

@@ -1,17 +1,17 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.2.0.M2</version>
+		<version>1.2.2.BUILD-SNAPSHOT</version>
 	</parent>
 	<artifactId>spring-cloud-starter-stream-kafka</artifactId>
 	<description>Spring Cloud Starter Stream Kafka</description>
-	<url>http://projects.spring.io/spring-cloud</url>
+	<url>https://projects.spring.io/spring-cloud</url>
 	<organization>
 		<name>Pivotal Software, Inc.</name>
-		<url>http://www.spring.io</url>
+		<url>https://www.spring.io</url>
 	</organization>
 	<properties>
 		<main.basedir>${basedir}/../..</main.basedir>

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

@@ -1,17 +1,17 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.2.0.M2</version>
+		<version>1.2.2.BUILD-SNAPSHOT</version>
 	</parent>
 	<artifactId>spring-cloud-stream-binder-kafka-0.10.0-test</artifactId>
 	<description>Spring Cloud Stream Kafka Binder 0.10.0 Tests</description>
-	<url>http://projects.spring.io/spring-cloud</url>
+	<url>https://projects.spring.io/spring-cloud</url>
 	<organization>
 		<name>Pivotal Software, Inc.</name>
-		<url>http://www.spring.io</url>
+		<url>https://www.spring.io</url>
 	</organization>
 	<properties>
 		<main.basedir>${basedir}/../..</main.basedir>
@@ -28,7 +28,7 @@
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
-			<version>1.2.0.M2</version>
+			<version>1.2.2.BUILD-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
@@ -101,7 +101,7 @@
 	<repositories>
 		<repository>
 			<id>confluent</id>
-			<url>http://packages.confluent.io/maven/</url>
+			<url>https://packages.confluent.io/maven/</url>
 		</repository>
 	</repositories>
 

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,17 +1,17 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.2.0.M2</version>
+		<version>1.2.2.BUILD-SNAPSHOT</version>
 	</parent>
 	<artifactId>spring-cloud-stream-binder-kafka-0.9</artifactId>
 	<description>Spring Cloud Stream Kafka Binder 0.9 Tests</description>
-	<url>http://projects.spring.io/spring-cloud</url>
+	<url>https://projects.spring.io/spring-cloud</url>
 	<organization>
 		<name>Pivotal Software, Inc.</name>
-		<url>http://www.spring.io</url>
+		<url>https://www.spring.io</url>
 	</organization>
 	<properties>
 		<main.basedir>${basedir}/../..</main.basedir>

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,17 +1,17 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.2.0.M2</version>
+        <version>1.2.2.BUILD-SNAPSHOT</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
     <description>Spring Cloud Stream Kafka Binder Core</description>
-    <url>http://projects.spring.io/spring-cloud</url>
+    <url>https://projects.spring.io/spring-cloud</url>
     <organization>
         <name>Pivotal Software, Inc.</name>
-        <url>http://www.spring.io</url>
+        <url>https://www.spring.io</url>
     </organization>
     <properties>
 

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,11 +1,11 @@
 /*
- * 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.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://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,
@@ -27,13 +27,14 @@ 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 {
 
 	private String[] zkNodes = new String[] {"localhost"};
 
-	private Map<String, String> configuration = new HashMap<>();
+	private Map<String, Object> configuration = new HashMap<>();
 
 	private String defaultZkPort = "2181";
 
@@ -77,6 +78,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() {
@@ -217,6 +223,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;
 	}
@@ -249,11 +263,11 @@ public class KafkaBinderConfigurationProperties {
 		this.socketBufferSize = socketBufferSize;
 	}
 
-	public Map<String, String> getConfiguration() {
+	public Map<String, Object> getConfiguration() {
 		return configuration;
 	}
 
-	public void setConfiguration(Map<String, String> configuration) {
+	public void setConfiguration(Map<String, Object> configuration) {
 		this.configuration = configuration;
 	}
 

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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
+ *      https://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,
@@ -21,6 +21,7 @@ import java.util.Map;
 
 /**
  * @author Marius Bogoevici
+ * @author Ilayaperumal Gopinathan
  *
  * <p>Thanks to Laszlo Szabo for providing the initial patch for generic property support.</p>
  */
@@ -38,6 +39,8 @@ public class KafkaConsumerProperties {
 
 	private boolean enableDlq;
 
+	private String dlqName;
+
 	private int recoveryInterval = 5000;
 
 	private Map<String, String> configuration = new HashMap<>();
@@ -119,4 +122,12 @@ public class KafkaConsumerProperties {
 	public void setConfiguration(Map<String, String> configuration) {
 		this.configuration = configuration;
 	}
+
+	public String getDlqName() {
+		return dlqName;
+	}
+
+	public void setDlqName(String dlqName) {
+		this.dlqName = dlqName;
+	}
 }

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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
+ *      https://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,
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.kafka.properties;
 
+import org.springframework.expression.Expression;
+
 import java.util.HashMap;
 import java.util.Map;
 
@@ -23,6 +25,7 @@ import javax.validation.constraints.NotNull;
 
 /**
  * @author Marius Bogoevici
+ * @author Henryk Konsek
  */
 public class KafkaProducerProperties {
 
@@ -34,6 +37,8 @@ public class KafkaProducerProperties {
 
 	private int batchTimeout;
 
+	private Expression messageKeyExpression;
+
 	private Map<String, String> configuration = new HashMap<>();
 
 	public int getBufferSize() {
@@ -69,6 +74,14 @@ public class KafkaProducerProperties {
 		this.batchTimeout = batchTimeout;
 	}
 
+	public Expression getMessageKeyExpression() {
+		return messageKeyExpression;
+	}
+
+	public void setMessageKeyExpression(Expression messageKeyExpression) {
+		this.messageKeyExpression = messageKeyExpression;
+	}
+
 	public Map<String, String> getConfiguration() {
 		return this.configuration;
 	}

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -20,6 +20,8 @@ import java.util.Collection;
 import java.util.Properties;
 import java.util.concurrent.Callable;
 
+import kafka.common.ErrorMapping;
+import kafka.utils.ZkUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.PartitionInfo;
@@ -47,14 +49,13 @@ import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import kafka.common.ErrorMapping;
-import kafka.utils.ZkUtils;
-
 /**
  * Kafka implementation for {@link ProvisioningProvider}
  *
  * @author Soby Chacko
  * @author Gary Russell
+ * @author Ilayaperumal Gopinathan
+ * @author Simon Flandergan
  */
 public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<KafkaConsumerProperties>,
 		ExtendedProducerProperties<KafkaProducerProperties>>, InitializingBean {
@@ -74,7 +75,6 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	/**
-	 *
 	 * @param metadataRetryOperations the retry configuration
 	 */
 	public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) {
@@ -105,7 +105,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			this.logger.info("Using kafka topic for outbound: " + name);
 		}
 		KafkaTopicUtils.validateTopicName(name);
-		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount());
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount(), false);
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -129,7 +129,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
 		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
-		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount);
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount, properties.getExtension().isAutoRebalanceEnabled());
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -137,8 +137,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					JaasUtils.isZkSecurityEnabled());
 			int partitions = adminUtilsOperation.partitionSize(name, zkUtils);
 			if (properties.getExtension().isEnableDlq() && !anonymous) {
-				String dlqTopic = "error." + name + "." + group;
-				createTopicAndPartitions(dlqTopic, partitions);
+				String dlqTopic = StringUtils.hasText(properties.getExtension().getDlqName()) ?
+						properties.getExtension().getDlqName() : "error." + name + "." + group;
+				createTopicAndPartitions(dlqTopic, partitions, properties.getExtension().isAutoRebalanceEnabled());
 				return new KafkaConsumerDestination(name, partitions, dlqTopic);
 			}
 			return new KafkaConsumerDestination(name, partitions);
@@ -146,9 +147,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		return new KafkaConsumerDestination(name);
 	}
 
-	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount) {
+	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount,
+																	boolean tolerateLowerPartitionsOnBroker) {
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
-			createTopicAndPartitions(topicName, partitionCount);
+			createTopicAndPartitions(topicName, partitionCount, tolerateLowerPartitionsOnBroker);
 		}
 		else if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation == null) {
 			this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " +
@@ -163,7 +165,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	 * Creates a Kafka topic if needed, or try to increase its partition count to the
 	 * desired number.
 	 */
-	private void createTopicAndPartitions(final String topicName, final int partitionCount) {
+	private void createTopicAndPartitions(final String topicName, final int partitionCount,
+										boolean tolerateLowerPartitionsOnBroker) {
+
 		final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 				this.configurationProperties.getZkSessionTimeout(),
 				this.configurationProperties.getZkConnectionTimeout(),
@@ -181,6 +185,11 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					if (this.configurationProperties.isAutoAddPartitions()) {
 						adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
 					}
+					else if (tolerateLowerPartitionsOnBroker) {
+						logger.warn("The number of expected partitions was: " + partitionCount + ", but "
+								+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+								+ "There will be " + (effectivePartitionCount - partitionSize) + " idle consumers");
+					}
 					else {
 						throw new ProvisioningException("The number of expected partitions was: " + partitionCount + ", but "
 								+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
@@ -199,8 +208,22 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					@Override
 					public Object doWithRetry(RetryContext context) throws RuntimeException {
 
-						adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount,
-								configurationProperties.getReplicationFactor(), new Properties());
+						try {
+							adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount,
+									configurationProperties.getReplicationFactor(), new Properties());
+						}
+						catch (Exception e) {
+							String exceptionClass = e.getClass().getName();
+							if (exceptionClass.equals("kafka.common.TopicExistsException")
+									|| exceptionClass.equals("org.apache.kafka.common.errors.TopicExistsException")) {
+								if (logger.isWarnEnabled()) {
+									logger.warn("Attempt to create topic: " + topicName + ". Topic already exists.");
+								}
+							}
+							else {
+								throw e;
+							}
+						}
 						return null;
 					}
 				});
@@ -215,7 +238,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		}
 	}
 
-	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount, final Callable<Collection<PartitionInfo>> callable) {
+	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount,
+														final boolean tolerateLowerPartitionsOnBroker,
+														final Callable<Collection<PartitionInfo>> callable) {
 		try {
 			return this.metadataRetryOperations
 					.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
@@ -224,10 +249,18 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 						public Collection<PartitionInfo> doWithRetry(RetryContext context) throws Exception {
 							Collection<PartitionInfo> partitions = callable.call();
 							// do a sanity check on the partition set
-							if (partitions.size() < partitionCount) {
-								throw new IllegalStateException("The number of expected partitions was: "
-										+ partitionCount + ", but " + partitions.size()
-										+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
+							int partitionSize = partitions.size();
+							if (partitionSize < partitionCount) {
+								if (tolerateLowerPartitionsOnBroker) {
+									logger.warn("The number of expected partitions was: " + partitionCount + ", but "
+											+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+											+ "There will be " + (partitionCount - partitionSize) + " idle consumers");
+								}
+								else {
+									throw new IllegalStateException("The number of expected partitions was: "
+											+ partitionCount + ", but " + partitionSize
+											+ (partitionSize > 1 ? " have " : " has ") + "been found instead");
+								}
 							}
 							return partitions;
 						}
@@ -308,7 +341,5 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					", dlqName='" + dlqName + '\'' +
 					'}';
 		}
-
 	}
-
 }

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/KafkaTopicUtils.java

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,11 +1,11 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.2.0.M2</version>
+        <version>1.2.2.BUILD-SNAPSHOT</version>
     </parent>
 
     <artifactId>spring-cloud-stream-binder-kafka-docs</artifactId>
@@ -71,8 +71,8 @@
                                     <quiet>true</quiet>
                                     <stylesheetfile>${basedir}/src/main/javadoc/spring-javadoc.css</stylesheetfile>
                                     <links>
-                                        <link>http://docs.spring.io/spring-framework/docs/${spring.version}/javadoc-api/</link>
-                                        <link>http://docs.spring.io/spring-shell/docs/current/api/</link>
+                                        <link>https://docs.spring.io/spring-framework/docs/${spring.version}/javadoc-api/</link>
+                                        <link>https://docs.spring.io/spring-shell/docs/current/api/</link>
                                     </links>
                                 </configuration>
                             </execution>

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

@@ -34,7 +34,7 @@ source control.
 
 The projects that require middleware generally include a
 `docker-compose.yml`, so consider using
-http://compose.docker.io/[Docker Compose] to run the middeware servers
+https://compose.docker.io/[Docker Compose] to run the middeware servers
 in Docker containers.
 
 === Documentation
@@ -43,13 +43,13 @@ There is a "full" profile that will generate documentation.
 
 === Working with the code
 If you don't have an IDE preference we would recommend that you use
-http://www.springsource.com/developer/sts[Spring Tools Suite] or
-http://eclipse.org[Eclipse] when working with the code. We use the
-http://eclipse.org/m2e/[m2eclipe] eclipse plugin for maven support. Other IDEs and tools
+https://www.springsource.com/developer/sts[Spring Tools Suite] or
+https://eclipse.org[Eclipse] when working with the code. We use the
+https://eclipse.org/m2e/[m2eclipe] eclipse plugin for maven support. Other IDEs and tools
 should also work without issue.
 
 ==== Importing into eclipse with m2eclipse
-We recommend the http://eclipse.org/m2e/[m2eclipe] eclipse plugin when working with
+We recommend the https://eclipse.org/m2e/[m2eclipe] eclipse plugin when working with
 eclipse. If you don't already have m2eclipse installed it is available from the "eclipse
 marketplace".
 

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

@@ -24,7 +24,7 @@ added after the original pull request but before a merge.
   `eclipse-code-formatter.xml` file from the
   https://github.com/spring-cloud/build/tree/master/eclipse-coding-conventions.xml[Spring
   Cloud Build] project. If using IntelliJ, you can use the
-  http://plugins.jetbrains.com/plugin/6546[Eclipse Code Formatter
+  https://plugins.jetbrains.com/plugin/6546[Eclipse Code Formatter
   Plugin] to import the same file.
 * Make sure all new `.java` files to have a simple Javadoc class comment with at least an
   `@author` tag identifying you, and preferably at least a paragraph on what the class is
@@ -37,6 +37,6 @@ added after the original pull request but before a merge.
 * A few unit tests would help a lot as well -- someone has to do it.
 * If no-one else is using your branch, please rebase it against the current master (or
   other target branch in the main project).
-* When writing a commit message please follow http://tbaggery.com/2008/04/19/a-note-about-git-commit-messages.html[these conventions],
+* When writing a commit message please follow https://tbaggery.com/2008/04/19/a-note-about-git-commit-messages.html[these conventions],
   if you are fixing an existing issue please add `Fixes gh-XXXX` at the end of the commit
   message (where XXXX is the issue number).

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

@@ -0,0 +1,109 @@
+[[kafka-dlq-processing]]
+== Dead-Letter Topic Processing
+
+Because it can't be anticipated how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them.
+If the reason for the dead-lettering is transient, you may wish to route the messages back to the original topic.
+However, if the problem is a permanent issue, that could cause an infinite loop.
+The following `spring-boot` application is an example of how to route those messages back to the original topic, but moves them to a third "parking lot" topic after three attempts.
+The application is simply another spring-cloud-stream application that reads from the dead-letter topic.
+It terminates when no messages are received for 5 seconds.
+
+The examples assume the original destination is `so8400out` and the consumer group is `so8400`.
+
+There are several considerations.
+
+- Consider only running the rerouting when the main application is not running.
+Otherwise, the retries for transient errors will be used up very quickly.
+- Alternatively, use a two-stage approach - use this application to route to a third topic, and another to route from there back to the main topic.
+- Since this technique uses a message header to keep track of retries, it won't work with `headerMode=raw`.
+In that case, consider adding some data to the payload (that can be ignored by the main application).
+- `x-retries` has to be added to the `headers` property `spring.cloud.stream.kafka.binder.headers=x-retries` on both this, and the main application so that the header is transported between the applications.
+- Since kafka is publish/subscribe, replayed messages will be sent to each consumer group, even those that successfully processed a message the first time around.
+
+.application.properties
+[source]
+----
+spring.cloud.stream.bindings.input.group=so8400replay
+spring.cloud.stream.bindings.input.destination=error.so8400out.so8400
+
+spring.cloud.stream.bindings.output.destination=so8400out
+spring.cloud.stream.bindings.output.producer.partitioned=true
+
+spring.cloud.stream.bindings.parkingLot.destination=so8400in.parkingLot
+spring.cloud.stream.bindings.parkingLot.producer.partitioned=true
+
+spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=earliest
+
+spring.cloud.stream.kafka.binder.headers=x-retries
+----
+
+.Application
+[source, java]
+----
+@SpringBootApplication
+@EnableBinding(TwoOutputProcessor.class)
+public class ReRouteDlqKApplication implements CommandLineRunner {
+
+    private static final String X_RETRIES_HEADER = "x-retries";
+
+    public static void main(String[] args) {
+        SpringApplication.run(ReRouteDlqKApplication.class, args).close();
+    }
+
+    private final AtomicInteger processed = new AtomicInteger();
+
+    @Autowired
+    private MessageChannel parkingLot;
+
+    @StreamListener(Processor.INPUT)
+    @SendTo(Processor.OUTPUT)
+    public Message<?> reRoute(Message<?> failed) {
+        processed.incrementAndGet();
+        Integer retries = failed.getHeaders().get(X_RETRIES_HEADER, Integer.class);
+        if (retries == null) {
+            System.out.println("First retry for " + failed);
+            return MessageBuilder.fromMessage(failed)
+                    .setHeader(X_RETRIES_HEADER, new Integer(1))
+                    .setHeader(BinderHeaders.PARTITION_OVERRIDE,
+                            failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
+                    .build();
+        }
+        else if (retries.intValue() < 3) {
+            System.out.println("Another retry for " + failed);
+            return MessageBuilder.fromMessage(failed)
+                    .setHeader(X_RETRIES_HEADER, new Integer(retries.intValue() + 1))
+                    .setHeader(BinderHeaders.PARTITION_OVERRIDE,
+                            failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
+                    .build();
+        }
+        else {
+            System.out.println("Retries exhausted for " + failed);
+            parkingLot.send(MessageBuilder.fromMessage(failed)
+                    .setHeader(BinderHeaders.PARTITION_OVERRIDE,
+                            failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
+                    .build());
+        }
+        return null;
+    }
+
+    @Override
+    public void run(String... args) throws Exception {
+        while (true) {
+            int count = this.processed.get();
+            Thread.sleep(5000);
+            if (count == this.processed.get()) {
+                System.out.println("Idle, terminating");
+                return;
+            }
+        }
+    }
+
+    public interface TwoOutputProcessor extends Processor {
+
+        @Output("parkingLot")
+        MessageChannel parkingLot();
+
+    }
+
+}
+----

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

@@ -11,19 +11,19 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
 :spring-cloud-stream-binder-kafka-repo: snapshot
 :github-tag: master
 :spring-cloud-stream-binder-kafka-docs-version: current
-:spring-cloud-stream-binder-kafka-docs: http://docs.spring.io/spring-cloud-stream-binder-kafka/docs/{spring-cloud-stream-binder-kafka-docs-version}/reference
-:spring-cloud-stream-binder-kafka-docs-current: http://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current-SNAPSHOT/reference/html/
+:spring-cloud-stream-binder-kafka-docs: https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/{spring-cloud-stream-binder-kafka-docs-version}/reference
+:spring-cloud-stream-binder-kafka-docs-current: https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current-SNAPSHOT/reference/html/
 :github-repo: spring-cloud/spring-cloud-stream-binder-kafka
-:github-raw: http://raw.github.com/{github-repo}/{github-tag}
-:github-code: http://github.com/{github-repo}/tree/{github-tag}
-:github-wiki: http://github.com/{github-repo}/wiki
-:github-master-code: http://github.com/{github-repo}/tree/master
+:github-raw: https://raw.github.com/{github-repo}/{github-tag}
+:github-code: https://github.com/{github-repo}/tree/{github-tag}
+:github-wiki: https://github.com/{github-repo}/wiki
+:github-master-code: https://github.com/{github-repo}/tree/master
 :sc-ext: java
 // ======================================================================================
 
 = Reference Guide
 include::overview.adoc[]
-
+include::dlq.adoc[]
 
 = Appendices
 [appendix]

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

@@ -41,7 +41,7 @@ Partitioning also maps directly to Apache Kafka partitions as well.
 
 This section contains the configuration options used by the Apache Kafka binder.
 
-For common configuration options and properties pertaining to binder, refer to the https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc#configuration-options[core docs].
+For common configuration options and properties pertaining to binder, refer to the <<binding-properties,core documentation>>.
 
 === Kafka Binder Properties
 
@@ -72,6 +72,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`.
@@ -158,7 +163,8 @@ If the consumer group is set explicitly for the consumer 'binding' (via `spring.
 Default: null (equivalent to `earliest`).
 enableDlq::
   When set to true, it will send enable DLQ behavior for the consumer.
-  Messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`.
+  By default, messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`.
+  The DLQ topic name can be configurable via the property `dlqName`.
   This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
 +
 Default: `false`.
@@ -166,6 +172,10 @@ configuration::
   Map with a key/value pair containing generic Kafka consumer properties.
 +
 Default: Empty map.
+dlqName::
+  The name of the DLQ topic to receive the error messages.
++
+Default: null (If not specified, messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`).
 
 === Kafka Producer Properties
 
@@ -185,6 +195,11 @@ batchTimeout::
 (Normally the producer does not wait at all, and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency.
 +
 Default: `0`.
+messageKeyExpression::
+ A SpEL expression evaluated against the outgoing message used to populate the key of the produced Kafka message.
+For example `headers.key` or `payload.myKey`.
++
+Default: `none`.
 configuration::
   Map with a key/value pair containing generic Kafka producer properties.
 +
@@ -234,7 +249,7 @@ public class ManuallyAcknowdledgingConsumer {
 ==== Example: security configuration
 
 Apache Kafka 0.9 supports secure connections between client and brokers.
-To take advantage of this feature, follow the guidelines in the http://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 http://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation].
+To take advantage of this feature, follow the guidelines in the https://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 https://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation].
 Use the `spring.cloud.stream.kafka.binder.configuration` option to set security properties for all clients created by the binder.
 
 For example, for setting `security.protocol` to `SASL_SSL`, set:
@@ -246,7 +261,7 @@ spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL
 
 All the other security properties can be set in a similar manner.
 
-When using Kerberos, follow the instructions in the http://kafka.apache.org/090/documentation.html#security_sasl_clientconfig[reference documentation] for creating and referencing the JAAS configuration.
+When using Kerberos, follow the instructions in the https://kafka.apache.org/090/documentation.html#security_sasl_clientconfig[reference documentation] for creating and referencing the JAAS configuration.
 
 Spring Cloud Stream supports passing JAAS configuration information to the application using a JAAS configuration file and using Spring Boot properties.
 

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/common.xsl

@@ -9,7 +9,7 @@
  "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
+   https://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
@@ -20,7 +20,7 @@
 -->
 
 <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
-		xmlns:xslthl="http://xslthl.sf.net"
+		xmlns:xslthl="http://xslthl.sourceforge.net/"
 		xmlns:d="http://docbook.org/ns/docbook"
 		exclude-result-prefixes="xslthl d"
 		version='1.0'>

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/epub.xsl

@@ -9,7 +9,7 @@ to you 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
+https://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
@@ -20,7 +20,7 @@ under the License.
 -->
 
 <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
-		xmlns:xslthl="http://xslthl.sf.net"
+		xmlns:xslthl="http://xslthl.sourceforge.net/"
 		xmlns:d="http://docbook.org/ns/docbook"
 		exclude-result-prefixes="xslthl d"
 		version='1.0'>

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/html-multipage.xsl

@@ -9,7 +9,7 @@ to you 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
+https://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

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/html-singlepage.xsl

@@ -9,7 +9,7 @@ to you 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
+https://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

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/html.xsl

@@ -9,7 +9,7 @@ to you 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
+https://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
@@ -20,7 +20,7 @@ under the License.
 -->
 
 <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
-		xmlns:xslthl="http://xslthl.sf.net"
+		xmlns:xslthl="http://xslthl.sourceforge.net/"
 		xmlns:d="http://docbook.org/ns/docbook"
 		exclude-result-prefixes="xslthl"
 		version='1.0'>

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/pdf.xsl

@@ -9,7 +9,7 @@ to you 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
+https://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
@@ -22,7 +22,7 @@ under the License.
 <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
 				xmlns:d="http://docbook.org/ns/docbook"
 				xmlns:fo="http://www.w3.org/1999/XSL/Format"
-				xmlns:xslthl="http://xslthl.sf.net"
+				xmlns:xslthl="http://xslthl.sourceforge.net/"
 				xmlns:xlink='http://www.w3.org/1999/xlink'
 				xmlns:exsl="http://exslt.org/common"
 				exclude-result-prefixes="exsl xslthl d xlink"

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl-config.xml

@@ -19,5 +19,5 @@
 	<highlighter id="properties" file="./xslthl/properties-hl.xml" />
 	<highlighter id="json" file="./xslthl/json-hl.xml" />
 	<highlighter id="yaml" file="./xslthl/yaml-hl.xml" />
-	<namespace prefix="xslthl" uri="http://xslthl.sf.net" />
+	<namespace prefix="xslthl" uri="http://xslthl.sourceforge.net/" />
 </xslthl-config>

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/bourne-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for SH
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2010 Mathieu Malaterre
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/c-hl.xml

@@ -3,7 +3,7 @@
 Syntax highlighting definition for C
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/cpp-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for C++
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/csharp-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for C#
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/css-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for CSS files
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2011-2012 Martin Hujer, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied
@@ -26,7 +26,7 @@ freely, subject to the following restrictions:
 Martin Hujer <mhujer at users.sourceforge.net>
 Michiel Hendriks <elmuerte at users.sourceforge.net>
 
-Reference: http://www.w3.org/TR/CSS21/propidx.html
+Reference: https://www.w3.org/TR/CSS21/propidx.html
 
 -->
 <highlighters>

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/html-hl.xml

@@ -7,7 +7,7 @@
   myxml-hl.xml - konfigurace zvyraznovace XML, ktera zvlast zvyrazni
                  HTML elementy a XSL elementy
 
-  This file has been customized for the Asciidoctor project (http://asciidoctor.org).
+  This file has been customized for the Asciidoctor project (https://asciidoctor.org).
 -->
 <highlighters>
   <highlighter type="xml">

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/ini-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for ini files
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/java-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for Java
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/javascript-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for JavaScript
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/perl-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for Perl
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/php-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for PHP
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/properties-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for Java
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/python-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for Python
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/ruby-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for Ruby
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2005-2008 Michal Molhanec, Jirka Kosek, Michiel Hendriks
 
 This software is provided 'as-is', without any express or implied

spring-cloud-stream-binder-kafka-docs/src/main/docbook/xsl/xslthl/sql2003-hl.xml

@@ -4,7 +4,7 @@
 Syntax highlighting definition for SQL:1999
 
 xslthl - XSLT Syntax Highlighting
-http://sourceforge.net/projects/xslthl/
+https://sourceforge.net/projects/xslthl/
 Copyright (C) 2012 Michiel Hendriks, Martin Hujer, k42b3
 
 This software is provided 'as-is', without any express or implied

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

@@ -1,10 +1,10 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.2.0.M2</version>
+        <version>1.2.2.BUILD-SNAPSHOT</version>
     </parent>
     <artifactId>spring-cloud-stream-binder-kafka-test-support</artifactId>
     <description>Kafka related test classes</description>
@@ -19,6 +19,7 @@
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka-test</artifactId>
             <version>${spring-kafka.version}</version>
+            <scope>compile</scope>
         </dependency>
     </dependencies>
 </project>

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

@@ -1,5 +1,5 @@
 <?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">
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 
 	<artifactId>spring-cloud-stream-binder-kafka</artifactId>
@@ -10,14 +10,14 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.2.0.M2</version>
+		<version>1.2.2.BUILD-SNAPSHOT</version>
 	</parent>
 
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
-			<version>1.2.0.M2</version>
+			<version>1.2.2.BUILD-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
@@ -45,12 +45,6 @@
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka_2.11</artifactId>
-			<exclusions>
-				<exclusion>
-					<groupId>jline</groupId>
-					<artifactId>jline</artifactId>
-				</exclusion>
-			</exclusions>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
@@ -110,7 +104,7 @@
 	<repositories>
 		<repository>
 			<id>confluent</id>
-			<url>http://packages.confluent.io/maven/</url>
+			<url>https://packages.confluent.io/maven/</url>
 		</repository>
 	</repositories>
 

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -19,6 +19,9 @@ package org.springframework.cloud.stream.binder.kafka;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.env.EnvironmentPostProcessor;
 import org.springframework.core.env.ConfigurableEnvironment;
@@ -32,13 +35,34 @@ import org.springframework.core.env.MapPropertySource;
  */
 public class KafkaBinderEnvironmentPostProcessor implements EnvironmentPostProcessor {
 
+	public final static String SPRING_KAFKA = "spring.kafka";
+
+	public final static String SPRING_KAFKA_PRODUCER = SPRING_KAFKA + ".producer";
+
+	public final static String SPRING_KAFKA_CONSUMER = SPRING_KAFKA + ".consumer";
+
+	public final static String SPRING_KAFKA_PRODUCER_KEY_SERIALIZER = SPRING_KAFKA_PRODUCER + "." + "keySerializer";
+
+	public final static String SPRING_KAFKA_PRODUCER_VALUE_SERIALIZER = SPRING_KAFKA_PRODUCER + "." + "valueSerializer";
+
+	public final static String SPRING_KAFKA_CONSUMER_KEY_DESERIALIZER = SPRING_KAFKA_CONSUMER + "." + "keyDeserializer";
+
+	public final static String SPRING_KAFKA_CONSUMER_VALUE_DESERIALIZER = SPRING_KAFKA_CONSUMER + "." + "valueDeserializer";
+
+	private static final String KAFKA_BINDER_DEFAULT_PROPERTIES = "kafkaBinderDefaultProperties";
+
 	@Override
 	public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
-		Map<String, Object> propertiesToAdd = new HashMap<>();
-		propertiesToAdd.put("logging.pattern.console", "%d{ISO8601} %5p %t %c{2}:%L - %m%n");
-		propertiesToAdd.put("logging.level.org.I0Itec.zkclient", "ERROR");
-		propertiesToAdd.put("logging.level.kafka.server.KafkaConfig", "ERROR");
-		propertiesToAdd.put("logging.level.kafka.admin.AdminClient.AdminConfig", "ERROR");
-		environment.getPropertySources().addLast(new MapPropertySource("kafkaBinderLogConfig", propertiesToAdd));
+		if (!environment.getPropertySources().contains(KAFKA_BINDER_DEFAULT_PROPERTIES)) {
+			Map<String, Object> kafkaBinderDefaultProperties = new HashMap<>();
+			kafkaBinderDefaultProperties.put("logging.level.org.I0Itec.zkclient", "ERROR");
+			kafkaBinderDefaultProperties.put("logging.level.kafka.server.KafkaConfig", "ERROR");
+			kafkaBinderDefaultProperties.put("logging.level.kafka.admin.AdminClient.AdminConfig", "ERROR");
+			kafkaBinderDefaultProperties.put(SPRING_KAFKA_PRODUCER_KEY_SERIALIZER, ByteArraySerializer.class.getName());
+			kafkaBinderDefaultProperties.put(SPRING_KAFKA_PRODUCER_VALUE_SERIALIZER, ByteArraySerializer.class.getName());
+			kafkaBinderDefaultProperties.put(SPRING_KAFKA_CONSUMER_KEY_DESERIALIZER, ByteArrayDeserializer.class.getName());
+			kafkaBinderDefaultProperties.put(SPRING_KAFKA_CONSUMER_VALUE_DESERIALIZER, ByteArrayDeserializer.class.getName());
+			environment.getPropertySources().addLast(new MapPropertySource(KAFKA_BINDER_DEFAULT_PROPERTIES, kafkaBinderDefaultProperties));
+		}
 	}
 }

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -16,70 +16,109 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 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.ConsumerConfig;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.HealthIndicator;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.kafka.core.ConsumerFactory;
 
 /**
  * Health indicator for Kafka.
  *
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
 public class KafkaBinderHealthIndicator implements HealthIndicator {
 
+	private static final int DEFAULT_TIMEOUT = 60;
+
 	private final KafkaMessageChannelBinder binder;
 
-	private final KafkaBinderConfigurationProperties configurationProperties;
+	private final ConsumerFactory<?, ?> consumerFactory;
+
+	private int timeout = DEFAULT_TIMEOUT;
 
 	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
-										KafkaBinderConfigurationProperties configurationProperties) {
+			ConsumerFactory<?, ?> consumerFactory) {
 		this.binder = binder;
-		this.configurationProperties = configurationProperties;
+		this.consumerFactory = consumerFactory;
 
 	}
 
+	/**
+	 * 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() {
-		Map<String, String> properties = new HashMap<>();
-		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties
-				.getKafkaConnectionString());
-		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
-		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
-		KafkaConsumer metadataConsumer = new KafkaConsumer(properties);
-		try {
-			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).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).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 {
-			metadataConsumer.close();
+			exec.shutdownNow();
 		}
 	}
+
 }

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -1,11 +1,11 @@
 /*
- * 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.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://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,
@@ -21,12 +21,15 @@ 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;
@@ -83,6 +86,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * @author Gary Russell
  * @author Mark Fisher
  * @author Soby Chacko
+ * @author Henryk Konsek
+ * @author Doug Saus
  */
 public class KafkaMessageChannelBinder extends
 		AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,
@@ -98,7 +103,7 @@ public class KafkaMessageChannelBinder extends
 	private final Map<String, Collection<PartitionInfo>> topicsInUse = new HashMap<>();
 
 	public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
-									KafkaTopicProvisioner provisioningProvider) {
+			KafkaTopicProvisioner provisioningProvider) {
 		super(false, headersToMap(configurationProperties), provisioningProvider);
 		this.configurationProperties = configurationProperties;
 	}
@@ -143,14 +148,21 @@ public class KafkaMessageChannelBinder extends
 
 	@Override
 	protected MessageHandler createProducerMessageHandler(final ProducerDestination destination,
-															ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) 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(), partitions);
 		if (producerProperties.getPartitionCount() < partitions.size()) {
@@ -171,20 +183,27 @@ public class KafkaMessageChannelBinder extends
 	private DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
 		Map<String, Object> props = new HashMap<>();
-		if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
-			props.putAll(configurationProperties.getConfiguration());
-		}
-		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
 		props.put(ProducerConfig.RETRIES_CONFIG, 0);
-		props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize()));
 		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks()));
-		props.put(ProducerConfig.LINGER_MS_CONFIG,
-				String.valueOf(producerProperties.getExtension().getBatchTimeout()));
-		props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
-				producerProperties.getExtension().getCompressionType().toString());
+		if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
+			props.putAll(configurationProperties.getConfiguration());
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
+			props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) {
+			props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize()));
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) {
+			props.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(producerProperties.getExtension().getBatchTimeout()));
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) {
+			props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
+					producerProperties.getExtension().getCompressionType().toString());
+		}
 		if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) {
 			props.putAll(producerProperties.getExtension().getConfiguration());
 		}
@@ -194,38 +213,40 @@ public class KafkaMessageChannelBinder extends
 	@Override
 	@SuppressWarnings("unchecked")
 	protected MessageProducer createConsumerEndpoint(final ConsumerDestination destination, final String group,
-													ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
 
 		boolean anonymous = !StringUtils.hasText(group);
-		Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(),
+		Assert.isTrue(!anonymous || !extendedConsumerProperties.getExtension().isEnableDlq(),
 				"DLQ support is not available for anonymous subscriptions");
 		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
-		Map<String, Object> props = getConsumerConfig(anonymous, consumerGroup);
-		if (!ObjectUtils.isEmpty(properties.getExtension().getConfiguration())) {
-			props.putAll(properties.getExtension().getConfiguration());
-		}
-		final ConsumerFactory<?, ?> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
-		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
+		final ConsumerFactory<?, ?> consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, extendedConsumerProperties);
+		int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency();
 
 		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;
 
-		if (properties.getExtension().isAutoRebalanceEnabled() ||
-				properties.getInstanceCount() == 1) {
+		if (extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ||
+				extendedConsumerProperties.getInstanceCount() == 1) {
 			listenedPartitions = allPartitions;
 		}
 		else {
 			listenedPartitions = new ArrayList<>();
 			for (PartitionInfo partition : allPartitions) {
 				// divide partitions across modules
-				if ((partition.partition() % properties.getInstanceCount()) == properties.getInstanceIndex()) {
+				if ((partition.partition() % extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties.getInstanceIndex()) {
 					listenedPartitions.add(partition);
 				}
 			}
@@ -236,9 +257,9 @@ public class KafkaMessageChannelBinder extends
 		final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
 				listenedPartitions);
 		final ContainerProperties containerProperties =
-				anonymous || properties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(destination.getName())
-						: new ContainerProperties(topicPartitionInitialOffsets);
-		int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size());
+				anonymous || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ?
+						new ContainerProperties(destination.getName()) : new ContainerProperties(topicPartitionInitialOffsets);
+		int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
 		final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
 				new ConcurrentMessageListenerContainer(
 						consumerFactory, containerProperties) {
@@ -249,8 +270,8 @@ public class KafkaMessageChannelBinder extends
 					}
 				};
 		messageListenerContainer.setConcurrency(concurrency);
-		messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(properties));
-		if (!properties.getExtension().isAutoCommitOffset()) {
+		messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(extendedConsumerProperties));
+		if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) {
 			messageListenerContainer.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
 		}
 		if (this.logger.isDebugEnabled()) {
@@ -265,21 +286,25 @@ public class KafkaMessageChannelBinder extends
 				new KafkaMessageDrivenChannelAdapter<>(
 						messageListenerContainer);
 		kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
-		final RetryTemplate retryTemplate = buildRetryTemplate(properties);
+		final RetryTemplate retryTemplate = buildRetryTemplate(extendedConsumerProperties);
 		kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate);
-		if (properties.getExtension().isEnableDlq()) {
+		if (extendedConsumerProperties.getExtension().isEnableDlq()) {
 			DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = getProducerFactory(new ExtendedProducerProperties<>(new KafkaProducerProperties()));
 			final KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFactory);
 			messageListenerContainer.getContainerProperties().setErrorHandler(new ErrorHandler() {
 
 				@Override
 				public void handle(Exception thrownException, final ConsumerRecord message) {
+					if (logger.isWarnEnabled()) {
+						logger.warn("Sending to DLQ:" + message, thrownException);
+					}
 					final byte[] key = message.key() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) message.key()))
 							: null;
 					final byte[] payload = message.value() != null
 							? Utils.toArray(ByteBuffer.wrap((byte[]) message.value())) : null;
-					ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send("error." + destination.getName() + "." + group,
-							message.partition(), key, payload);
+					String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName()) ?
+							extendedConsumerProperties.getExtension().getDlqName() : "error." + destination.getName() + "." + group;
+					ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(dlqName, message.partition(), key, payload);
 					sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
 						StringBuilder sb = new StringBuilder().append(" a message with key='")
 								.append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'")
@@ -308,20 +333,30 @@ public class KafkaMessageChannelBinder extends
 		return kafkaMessageDrivenChannelAdapter;
 	}
 
-	private Map<String, Object> getConsumerConfig(boolean anonymous, String consumerGroup) {
+	private ConsumerFactory<?, ?> createKafkaConsumerFactory(boolean anonymous, String consumerGroup,
+			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
 		Map<String, Object> props = new HashMap<>();
 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest");
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
+
 		if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
 			props.putAll(configurationProperties.getConfiguration());
 		}
-		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
-		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
-		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
-		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
-				anonymous ? "latest" : "earliest");
-		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
-		return props;
+		if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
+			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
+		}
+		if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getConfiguration())) {
+			props.putAll(consumerProperties.getExtension().getConfiguration());
+		}
+		if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) {
+			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumerProperties.getExtension().getStartOffset().name());
+		}
+
+		return new DefaultKafkaConsumerFactory<>(props);
 	}
 
 	private boolean isAutoCommitOnError(ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
@@ -358,10 +393,11 @@ public class KafkaMessageChannelBinder extends
 		private final DefaultKafkaProducerFactory<byte[], byte[]> producerFactory;
 
 		private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
-													ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
-													DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
+				ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
+				DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
 			super(kafkaTemplate);
 			setTopicExpression(new LiteralExpression(topic));
+			setMessageKeyExpression(producerProperties.getExtension().getMessageKeyExpression());
 			setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
 			if (producerProperties.isPartitioned()) {
 				SpelExpressionParser parser = new SpelExpressionParser();

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -16,16 +16,24 @@
 
 package org.springframework.cloud.stream.binder.kafka.config;
 
+import javax.annotation.PostConstruct;
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.utils.AppInfoParser;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
@@ -49,8 +57,11 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.core.type.AnnotatedTypeMetadata;
 import org.springframework.integration.codec.Codec;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
+import org.springframework.util.ObjectUtils;
 
 /**
  * @author David Turanski
@@ -58,10 +69,11 @@ import org.springframework.kafka.support.ProducerListener;
  * @author Soby Chacko
  * @author Mark Fisher
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
-@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
+@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, KafkaBinderConfiguration.KafkaPropertiesConfiguration.class})
 @EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class})
 public class KafkaBinderConfiguration {
 
@@ -108,7 +120,20 @@ public class KafkaBinderConfiguration {
 
 	@Bean
 	KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
-		return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, this.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);
+		if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
+			props.putAll(configurationProperties.getConfiguration());
+		}
+		if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
+			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
+		}
+		ConsumerFactory<?, ?> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
+		KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator(kafkaMessageChannelBinder,
+				consumerFactory);
+		indicator.setTimeout(this.configurationProperties.getHealthTimeout());
+		return indicator;
 	}
 
 	@Bean(name = "adminUtilsOperation")
@@ -154,4 +179,47 @@ public class KafkaBinderConfiguration {
 
 		private JaasLoginModuleConfiguration zookeeper;
 	}
+
+	@ConditionalOnClass(name = "org.springframework.boot.autoconfigure.kafka.KafkaProperties")
+	public static class KafkaPropertiesConfiguration {
+
+		// KafkaProperties can still be unavailable if KafkaAutoConfiguration is disabled.
+		@Autowired(required = false)
+		private KafkaProperties kafkaProperties;
+
+		@Autowired
+		private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;
+
+		@PostConstruct
+		public void init() {
+			Map<String, Object> configuration = this.kafkaBinderConfigurationProperties.getConfiguration();
+			if (this.kafkaProperties != null) {
+				for (Map.Entry<String, String> properties : this.kafkaProperties.getProperties().entrySet()) {
+					if (!configuration.containsKey(properties.getKey())) {
+						configuration.put(properties.getKey(), properties.getValue());
+					}
+				}
+				for (Map.Entry<String, Object> producerProperties : this.kafkaProperties.buildProducerProperties().entrySet()) {
+					if (!configuration.containsKey(producerProperties.getKey())) {
+						configuration.put(producerProperties.getKey(), producerProperties.getValue());
+					}
+				}
+				for (Map.Entry<String, Object> consumerProperties : this.kafkaProperties.buildConsumerProperties().entrySet()) {
+					if (!configuration.containsKey(consumerProperties.getKey())) {
+						configuration.put(consumerProperties.getKey(), consumerProperties.getValue());
+					}
+				}
+				if (ObjectUtils.isEmpty(configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
+					configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBinderConfigurationProperties.getKafkaConnectionString());
+				}
+				else {
+					@SuppressWarnings("unchecked")
+					List<String> bootStrapServers = (List<String>) configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
+					if (bootStrapServers.size() == 1 && bootStrapServers.get(0).equals("localhost:9092")) {
+						configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBinderConfigurationProperties.getKafkaConnectionString());
+					}
+				}
+			}
+		}
+	}
 }

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -188,7 +188,7 @@ public class Kafka10BinderTests extends KafkaBinderTests {
 		final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
 		Map<String, Object> schemaRegistryProps = new HashMap<>();
 		schemaRegistryProps.put("kafkastore.connection.url", configurationProperties.getZkConnectionString());
-		schemaRegistryProps.put("listeners", "http://0.0.0.0:8082");
+		schemaRegistryProps.put("listeners", "https://0.0.0.0:8082");
 		schemaRegistryProps.put("port", "8082");
 		schemaRegistryProps.put("kafkastore.topic", "_schemas");
 		SchemaRegistryConfig config = new SchemaRegistryConfig(schemaRegistryProps);

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2016-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.ReflectionUtils;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = {KafkaBinderAutoConfigurationPropertiesTest.KafkaBinderConfigProperties.class, KafkaBinderConfiguration.class})
+@TestPropertySource(locations = "classpath:binder-config-autoconfig.properties")
+public class KafkaBinderAutoConfigurationPropertiesTest {
+
+	@Autowired
+	private KafkaMessageChannelBinder kafkaMessageChannelBinder;
+
+	@Autowired
+	private KafkaBinderHealthIndicator kafkaBinderHealthIndicator;
+
+	@Test
+	public void testKafkaBinderConfigurationWithKafkaProperties() throws Exception {
+		assertNotNull(this.kafkaMessageChannelBinder);
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(new KafkaProducerProperties());
+		Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class);
+		getProducerFactoryMethod.setAccessible(true);
+		DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties);
+		Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class);
+		ReflectionUtils.makeAccessible(producerFactoryConfigField);
+		Map<String, Object> producerConfigs = (Map<String, Object>) ReflectionUtils.getField(producerFactoryConfigField, producerFactory);
+		assertTrue(producerConfigs.get("batch.size").equals(10));
+		assertTrue(producerConfigs.get("key.serializer").equals(LongSerializer.class));
+		assertTrue(producerConfigs.get("value.serializer").equals(LongSerializer.class));
+		assertTrue(producerConfigs.get("compression.type").equals("snappy"));
+		List<String> bootstrapServers = new ArrayList<>();
+		bootstrapServers.add("10.98.09.199:9092");
+		bootstrapServers.add("10.98.09.196:9092");
+		assertTrue((((List<String>) producerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers)));
+		Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class);
+		createKafkaConsumerFactoryMethod.setAccessible(true);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties());
+		DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties);
+		Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class);
+		ReflectionUtils.makeAccessible(consumerFactoryConfigField);
+		Map<String, Object> consumerConfigs = (Map<String, Object>) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory);
+		assertTrue(consumerConfigs.get("key.deserializer").equals(LongDeserializer.class));
+		assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class));
+		assertTrue(consumerConfigs.get("group.id").equals("groupIdFromBootConfig"));
+		assertTrue(consumerConfigs.get("auto.offset.reset").equals("earliest"));
+		assertTrue((((List<String>) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers)));
+	}
+
+	@Test
+	public void testKafkaHealthIndicatorProperties() {
+		assertNotNull(this.kafkaBinderHealthIndicator);
+		Field consumerFactoryField = ReflectionUtils.findField(KafkaBinderHealthIndicator.class, "consumerFactory",
+				ConsumerFactory.class);
+		ReflectionUtils.makeAccessible(consumerFactoryField);
+		DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) ReflectionUtils.getField(
+				consumerFactoryField, this.kafkaBinderHealthIndicator);
+		Field configField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class);
+		ReflectionUtils.makeAccessible(configField);
+		Map<String, Object> configs = (Map<String, Object>) ReflectionUtils.getField(configField, consumerFactory);
+		assertTrue(configs.containsKey("bootstrap.servers"));
+		List<String> bootstrapServers = new ArrayList<>();
+		bootstrapServers.add("10.98.09.199:9092");
+		bootstrapServers.add("10.98.09.196:9092");
+		assertTrue(((List<String>)configs.get("bootstrap.servers")).containsAll(bootstrapServers));
+	}
+
+	public static class KafkaBinderConfigProperties {
+
+		@Bean
+		KafkaProperties kafkaProperties() {
+			return new KafkaProperties();
+		}
+	}
+}

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2016-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.ReflectionUtils;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = {KafkaBinderConfiguration.class})
+@TestPropertySource(locations = "classpath:binder-config.properties")
+public class KafkaBinderConfigurationPropertiesTest {
+
+	@Autowired
+	private KafkaMessageChannelBinder kafkaMessageChannelBinder;
+
+	@Test
+	public void testKafkaBinderConfigurationProperties() throws Exception {
+		assertNotNull(this.kafkaMessageChannelBinder);
+		KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
+		kafkaProducerProperties.setBufferSize(12345);
+		kafkaProducerProperties.setBatchTimeout(100);
+		kafkaProducerProperties.setCompressionType(KafkaProducerProperties.CompressionType.gzip);
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(kafkaProducerProperties);
+		Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class);
+		getProducerFactoryMethod.setAccessible(true);
+		DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties);
+		Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class);
+		ReflectionUtils.makeAccessible(producerFactoryConfigField);
+		Map<String, Object> producerConfigs = (Map<String, Object>) ReflectionUtils.getField(producerFactoryConfigField, producerFactory);
+		assertTrue(producerConfigs.get("batch.size").equals("12345"));
+		assertTrue(producerConfigs.get("linger.ms").equals("100"));
+		assertTrue(producerConfigs.get("key.serializer").equals(ByteArraySerializer.class));
+		assertTrue(producerConfigs.get("value.serializer").equals(ByteArraySerializer.class));
+		assertTrue(producerConfigs.get("compression.type").equals("gzip"));
+		List<String> bootstrapServers = new ArrayList<>();
+		bootstrapServers.add("10.98.09.199:9082");
+		assertTrue((((String) producerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082")));
+		Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class);
+		createKafkaConsumerFactoryMethod.setAccessible(true);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties());
+		DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties);
+		Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class);
+		ReflectionUtils.makeAccessible(consumerFactoryConfigField);
+		Map<String, Object> consumerConfigs = (Map<String, Object>) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory);
+		assertTrue(consumerConfigs.get("key.deserializer").equals(ByteArrayDeserializer.class));
+		assertTrue(consumerConfigs.get("value.deserializer").equals(ByteArrayDeserializer.class));
+		assertTrue((((String) consumerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082")));
+	}
+}

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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
+ *      https://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,
@@ -15,8 +15,6 @@
  */
 package org.springframework.cloud.stream.binder.kafka;
 
-import static org.junit.Assert.assertNotNull;
-
 import java.lang.reflect.Field;
 
 import org.junit.Test;
@@ -29,6 +27,8 @@ import org.springframework.kafka.support.ProducerListener;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
 
+import static org.junit.Assert.assertNotNull;
+
 /**
  * @author Ilayaperumal Gopinathan
  */

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

@@ -0,0 +1,113 @@
+/*
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka;
+
+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.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+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;
+
+/**
+ * @author Barry Commins
+ * @author Gary Russell
+ */
+public class KafkaBinderHealthIndicatorTest {
+
+	private static final String TEST_TOPIC = "test";
+	private KafkaBinderHealthIndicator indicator;
+
+	@Mock
+	private DefaultKafkaConsumerFactory consumerFactory;
+
+	@Mock
+	private KafkaConsumer consumer;
+
+	@Mock
+	private KafkaMessageChannelBinder binder;
+
+	private Map<String, Collection<PartitionInfo>> topicsInUse = new HashMap<>();
+
+	@Before
+	public void setup() {
+		MockitoAnnotations.initMocks(this);
+		given(consumerFactory.createConsumer()).willReturn(consumer);
+		given(binder.getTopicsInUse()).willReturn(topicsInUse);
+		this.indicator = new KafkaBinderHealthIndicator(binder, consumerFactory);
+		this.indicator.setTimeout(10);
+	}
+
+	@Test
+	public void kafkaBinderIsUp() {
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, partitions);
+		given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+		verify(this.consumer).close();
+	}
+
+	@Test
+	public void kafkaBinderIsDown() {
+		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
+		topicsInUse.put(TEST_TOPIC, partitions);
+		given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
+		Health health = indicator.health();
+		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, 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));
+		return partitions;
+	}
+}

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,
@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka;
 import javax.security.auth.login.AppConfigurationEntry;
 
 import com.sun.security.auth.login.ConfigFile;
+
 import org.apache.kafka.common.security.JaasUtils;
 import org.junit.Test;
 

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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
+ *      https://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,
@@ -32,9 +32,12 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.assertj.core.api.Condition;
 import org.junit.Ignore;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.cloud.stream.binder.Binder;
@@ -81,9 +84,13 @@ import static org.junit.Assert.assertTrue;
 /**
  * @author Soby Chacko
  * @author Ilayaperumal Gopinathan
+ * @author Henryk Konsek
  */
 public abstract class KafkaBinderTests extends PartitionCapableBinderTests<AbstractKafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>,
-						ExtendedProducerProperties<KafkaProducerProperties>> {
+		ExtendedProducerProperties<KafkaProducerProperties>> {
+
+	@Rule
+	public ExpectedException expectedProvisioningException = ExpectedException.none();
 
 	@Override
 	protected ExtendedConsumerProperties<KafkaConsumerProperties> createConsumerProperties() {
@@ -113,10 +120,10 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 	protected abstract ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties);
 
 	protected abstract void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions,
-												int replicationFactor, Properties topicConfig);
+											int replicationFactor, Properties topicConfig);
 
 	protected abstract int invokePartitionSize(String topic,
-												ZkUtils zkUtils);
+											ZkUtils zkUtils);
 
 	@Test
 	@SuppressWarnings("unchecked")
@@ -274,6 +281,68 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		producerBinding.unbind();
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testConfigurableDlqName() throws Exception {
+		Binder binder = getBinder();
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		DirectChannel moduleInputChannel = new DirectChannel();
+		FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler();
+		moduleInputChannel.subscribe(handler);
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.setPartitionCount(10);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.setMaxAttempts(3);
+		consumerProperties.setBackOffInitialInterval(100);
+		consumerProperties.setBackOffMaxInterval(150);
+		consumerProperties.getExtension().setEnableDlq(true);
+		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+		String dlqName = "dlqTest";
+		consumerProperties.getExtension().setDlqName(dlqName);
+		long uniqueBindingId = System.currentTimeMillis();
+		Binding<MessageChannel> producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0",
+				moduleOutputChannel, producerProperties);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0",
+				"testGroup", moduleInputChannel, consumerProperties);
+		ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
+		dlqConsumerProperties.setMaxAttempts(1);
+		QueueChannel dlqChannel = new QueueChannel();
+		Binding<MessageChannel> dlqConsumerBinding = binder.bindConsumer(dlqName, null, dlqChannel, dlqConsumerProperties);
+
+		String testMessagePayload = "test." + UUID.randomUUID().toString();
+		Message<String> testMessage = MessageBuilder.withPayload(testMessagePayload).build();
+		moduleOutputChannel.send(testMessage);
+
+		Message<?> dlqMessage = receive(dlqChannel, 3);
+		assertThat(dlqMessage).isNotNull();
+		assertThat(dlqMessage.getPayload()).isEqualTo(testMessagePayload);
+
+		// first attempt fails
+		assertThat(handler.getReceivedMessages().entrySet()).hasSize(1);
+		Message<?> handledMessage = handler.getReceivedMessages().entrySet().iterator().next().getValue();
+		assertThat(handledMessage).isNotNull();
+		assertThat(handledMessage.getPayload()).isEqualTo(testMessagePayload);
+		assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
+		binderBindUnbindLatency();
+		dlqConsumerBinding.unbind();
+		consumerBinding.unbind();
+
+		// on the second attempt the message is not redelivered because the DLQ is set
+		QueueChannel successfulInputChannel = new QueueChannel();
+		consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup",
+				successfulInputChannel, consumerProperties);
+		String testMessage2Payload = "test." + UUID.randomUUID().toString();
+		Message<String> testMessage2 = MessageBuilder.withPayload(testMessage2Payload).build();
+		moduleOutputChannel.send(testMessage2);
+
+		Message<?> receivedMessage = receive(successfulInputChannel);
+		assertThat(receivedMessage.getPayload()).isEqualTo(testMessage2Payload);
+
+		binderBindUnbindLatency();
+		consumerBinding.unbind();
+		producerBinding.unbind();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testAutoCreateTopicsEnabledSucceeds() throws Exception {
@@ -312,10 +381,10 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 			producerProperties.getExtension().setCompressionType(
 					KafkaProducerProperties.CompressionType.valueOf(codec.toString()));
-			Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel,
+			Binding<MessageChannel> producerBinding = binder.bindProducer("testCompression", moduleOutputChannel,
 					producerProperties);
 			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
-			Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
+			Binding<MessageChannel> consumerBinding = binder.bindConsumer("testCompression", "test", moduleInputChannel,
 					consumerProperties);
 			Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload)
 					.build();
@@ -402,6 +471,35 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		consumerBinding.unbind();
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDynamicKeyExpression() throws Exception {
+		Binder binder = getBinder(createConfigurationProperties());
+		QueueChannel moduleInputChannel = new QueueChannel();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.getExtension().getConfiguration().put("key.serializer", StringSerializer.class.getName());
+		producerProperties.getExtension().setMessageKeyExpression(spelExpressionParser.parseExpression("headers.key"));
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		String uniqueBindingId = UUID.randomUUID().toString();
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				createProducerBindingProperties(producerProperties));
+		Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0",
+				moduleOutputChannel, producerProperties);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null,
+				moduleInputChannel, consumerProperties);
+		Thread.sleep(1000);
+		Message<?> message = MessageBuilder.withPayload("somePayload").setHeader("key", "myDynamicKey").build();
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = receive(moduleInputChannel);
+		assertThat(inbound).isNotNull();
+		String receivedKey = new String(inbound.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, byte[].class));
+		assertThat(receivedKey).isEqualTo("myDynamicKey");
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception {
@@ -670,13 +768,13 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 				createProducerBindingProperties(createProducerProperties()));
 		QueueChannel moduleInputChannel = new QueueChannel();
 
-		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.x", moduleOutputChannel,
+		Binding<MessageChannel> producerBinding = binder.bindProducer("testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff", moduleOutputChannel,
 				createProducerProperties());
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.getExtension().setAutoCommitOffset(false);
 
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.x", "test", moduleInputChannel,
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff", "test", moduleInputChannel,
 				consumerProperties);
 
 		String testPayload1 = "foo" + UUID.randomUUID().toString();
@@ -712,12 +810,12 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 				createProducerBindingProperties(createProducerProperties()));
 		QueueChannel moduleInputChannel = new QueueChannel();
 
-		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.x", moduleOutputChannel,
+		Binding<MessageChannel> producerBinding = binder.bindProducer("testManualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder", moduleOutputChannel,
 				createProducerProperties());
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.x", "test", moduleInputChannel,
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("testManualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder", "test", moduleInputChannel,
 				consumerProperties);
 
 		String testPayload1 = "foo" + UUID.randomUUID().toString();
@@ -1077,7 +1175,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception {
+	public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled() throws Exception {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
 
 		final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
@@ -1092,26 +1190,46 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		Binder binder = getBinder(configurationProperties);
 		GenericApplicationContext context = new GenericApplicationContext();
 		context.refresh();
-		//		binder.setApplicationContext(context);
-		//		binder.afterPropertiesSet();
+
 		DirectChannel output = new DirectChannel();
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		// this consumer must consume from partition 2
 		consumerProperties.setInstanceCount(3);
 		consumerProperties.setInstanceIndex(2);
-		Binding binding = null;
-		try {
-			binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(ProvisioningException.class);
-			assertThat(e)
-					.hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead");
-		}
-		finally {
-			if (binding != null) {
-				binding.unbind();
-			}
+		Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
+		binding.unbind();
+		assertThat(invokePartitionSize(testTopicName, zkUtils)).isEqualTo(1);
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled() throws Exception {
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+
+		final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
+				configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+
+		final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+
+		String testTopicName = "existing" + System.currentTimeMillis();
+		invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties());
+		configurationProperties.setAutoAddPartitions(false);
+		Binder binder = getBinder(configurationProperties);
+		GenericApplicationContext context = new GenericApplicationContext();
+		context.refresh();
+
+		DirectChannel output = new DirectChannel();
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		// this consumer must consume from partition 2
+		consumerProperties.setInstanceCount(3);
+		consumerProperties.setInstanceIndex(2);
+		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+		expectedProvisioningException.expect(ProvisioningException.class);
+		expectedProvisioningException.expectMessage("The number of expected partitions was: 3, but 1 has been found instead");
+		Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
+		if (binding != null) {
+			binding.unbind();
 		}
 	}
 
@@ -1224,7 +1342,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		Binding<?> binding = null;
 		try {
 			KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
-			Map<String, String> propertiesToOverride = configurationProperties.getConfiguration();
+			Map<String, Object> propertiesToOverride = configurationProperties.getConfiguration();
 			propertiesToOverride.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 			propertiesToOverride.put("value.deserializer", "org.apache.kafka.common.serialization.LongDeserializer");
 			configurationProperties.setConfiguration(propertiesToOverride);
@@ -1248,11 +1366,11 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 	}
 
 	private KafkaConsumer getKafkaConsumer(Binding binding) {
-		DirectFieldAccessor bindingAccessor = new DirectFieldAccessor((DefaultBinding)binding);
+		DirectFieldAccessor bindingAccessor = new DirectFieldAccessor((DefaultBinding) binding);
 		KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor.getPropertyValue("lifecycle");
 		DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
 		ConcurrentMessageListenerContainer messageListenerContainer = (ConcurrentMessageListenerContainer) adapterAccessor.getPropertyValue("messageListenerContainer");
-		DirectFieldAccessor containerAccessor = new DirectFieldAccessor((ConcurrentMessageListenerContainer)messageListenerContainer);
+		DirectFieldAccessor containerAccessor = new DirectFieldAccessor((ConcurrentMessageListenerContainer) messageListenerContainer);
 		DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor.getPropertyValue("consumerFactory");
 		return (KafkaConsumer) consumerFactory.createConsumer();
 	}
@@ -1358,7 +1476,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 
 		DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties));
 		output.setBeanName("test.output");
-		Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, properties);
+		Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.raw.0", output, properties);
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setConcurrency(2);
@@ -1369,19 +1487,19 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
 		QueueChannel input0 = new QueueChannel();
 		input0.setBeanName("test.input0J");
-		Binding<MessageChannel> input0Binding = binder.bindConsumer("partJ.0", "test", input0, consumerProperties);
+		Binding<MessageChannel> input0Binding = binder.bindConsumer("partJ.raw.0", "test", input0, consumerProperties);
 		consumerProperties.setInstanceIndex(1);
 		QueueChannel input1 = new QueueChannel();
 		input1.setBeanName("test.input1J");
-		Binding<MessageChannel> input1Binding = binder.bindConsumer("partJ.0", "test", input1, consumerProperties);
+		Binding<MessageChannel> input1Binding = binder.bindConsumer("partJ.raw.0", "test", input1, consumerProperties);
 		consumerProperties.setInstanceIndex(2);
 		QueueChannel input2 = new QueueChannel();
 		input2.setBeanName("test.input2J");
-		Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties);
+		Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.raw.0", "test", input2, consumerProperties);
 
-		output.send(new GenericMessage<>(new byte[] {(byte) 0}));
-		output.send(new GenericMessage<>(new byte[] {(byte) 1}));
-		output.send(new GenericMessage<>(new byte[] {(byte) 2}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 0}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 1}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 2}));
 
 		Message<?> receive0 = receive(input0);
 		assertThat(receive0).isNotNull();
@@ -1411,11 +1529,11 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 
 		DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties));
 		output.setBeanName("test.output");
-		Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, properties);
+		Binding<MessageChannel> outputBinding = binder.bindProducer("part.raw.0", output, properties);
 		try {
 			Object endpoint = extractEndpoint(outputBinding);
 			assertThat(getEndpointRouting(endpoint))
-					.contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']");
+					.contains(getExpectedRoutingBaseDestination("part.raw.0", "test") + "-' + headers['partition']");
 		}
 		catch (UnsupportedOperationException ignored) {
 		}
@@ -1430,23 +1548,23 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
 		QueueChannel input0 = new QueueChannel();
 		input0.setBeanName("test.input0S");
-		Binding<MessageChannel> input0Binding = binder.bindConsumer("part.0", "test", input0, consumerProperties);
+		Binding<MessageChannel> input0Binding = binder.bindConsumer("part.raw.0", "test", input0, consumerProperties);
 		consumerProperties.setInstanceIndex(1);
 		QueueChannel input1 = new QueueChannel();
 		input1.setBeanName("test.input1S");
-		Binding<MessageChannel> input1Binding = binder.bindConsumer("part.0", "test", input1, consumerProperties);
+		Binding<MessageChannel> input1Binding = binder.bindConsumer("part.raw.0", "test", input1, consumerProperties);
 		consumerProperties.setInstanceIndex(2);
 		QueueChannel input2 = new QueueChannel();
 		input2.setBeanName("test.input2S");
-		Binding<MessageChannel> input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties);
+		Binding<MessageChannel> input2Binding = binder.bindConsumer("part.raw.0", "test", input2, consumerProperties);
 
-		Message<byte[]> message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[] {2})
+		Message<byte[]> message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[]{2})
 				.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "kafkaBinderTestCommonsDelegate")
 				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
 				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
 		output.send(message2);
-		output.send(new GenericMessage<>(new byte[] {1}));
-		output.send(new GenericMessage<>(new byte[] {0}));
+		output.send(new GenericMessage<>(new byte[]{1}));
+		output.send(new GenericMessage<>(new byte[]{0}));
 		Message<?> receive0 = receive(input0);
 		assertThat(receive0).isNotNull();
 		Message<?> receive1 = receive(input1);
@@ -1469,19 +1587,44 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		QueueChannel moduleInputChannel = new QueueChannel();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setHeaderMode(HeaderMode.raw);
-		Binding<MessageChannel> producerBinding = binder.bindProducer("0", moduleOutputChannel,
+		Binding<MessageChannel> producerBinding = binder.bindProducer("raw.0", moduleOutputChannel,
 				producerProperties);
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setHeaderMode(HeaderMode.raw);
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("0", "test", moduleInputChannel,
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("raw.0", "test", moduleInputChannel,
 				consumerProperties);
-		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("kafkaBinderTestCommonsDelegate".getBytes()).build();
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("testSendAndReceiveWithRawMode".getBytes()).build();
 		// Let the consumer actually bind to the producer before sending a msg
 		binderBindUnbindLatency();
 		moduleOutputChannel.send(message);
 		Message<?> inbound = receive(moduleInputChannel);
 		assertThat(inbound).isNotNull();
-		assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("kafkaBinderTestCommonsDelegate");
+		assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("testSendAndReceiveWithRawMode");
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testSendAndReceiveWithRawModeAndStringPayload() throws Exception {
+		Binder binder = getBinder();
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		QueueChannel moduleInputChannel = new QueueChannel();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.setHeaderMode(HeaderMode.raw);
+		Binding<MessageChannel> producerBinding = binder.bindProducer("raw.string.0", moduleOutputChannel,
+				producerProperties);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.setHeaderMode(HeaderMode.raw);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("raw.string.0", "test", moduleInputChannel,
+				consumerProperties);
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("testSendAndReceiveWithRawModeAndStringPayload").build();
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = receive(moduleInputChannel);
+		assertThat(inbound).isNotNull();
+		assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("testSendAndReceiveWithRawModeAndStringPayload");
 		producerBinding.unbind();
 		consumerBinding.unbind();
 	}
@@ -1497,30 +1640,30 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		QueueChannel module3InputChannel = new QueueChannel();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setHeaderMode(HeaderMode.raw);
-		Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel,
+		Binding<MessageChannel> producerBinding = binder.bindProducer("baz.raw.0", moduleOutputChannel,
 				producerProperties);
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setHeaderMode(HeaderMode.raw);
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
-		Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel,
+		Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.raw.0", "test", module1InputChannel,
 				consumerProperties);
 		// A new module is using the tap as an input channel
-		String fooTapName = "baz.0";
+		String fooTapName = "baz.raw.0";
 		Binding<MessageChannel> input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel,
 				consumerProperties);
 		// Another new module is using tap as an input channel
-		String barTapName = "baz.0";
+		String barTapName = "baz.raw.0";
 		Binding<MessageChannel> input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel,
 				consumerProperties);
 
-		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("kafkaBinderTestCommonsDelegate".getBytes()).build();
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("testSendAndReceiveWithExplicitConsumerGroupWithRawMode".getBytes()).build();
 		boolean success = false;
 		boolean retried = false;
 		while (!success) {
 			moduleOutputChannel.send(message);
 			Message<?> inbound = receive(module1InputChannel);
 			assertThat(inbound).isNotNull();
-			assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("kafkaBinderTestCommonsDelegate");
+			assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode");
 
 			Message<?> tapped1 = receive(module2InputChannel);
 			Message<?> tapped2 = receive(module3InputChannel);
@@ -1531,8 +1674,8 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 				continue;
 			}
 			success = true;
-			assertThat(new String((byte[]) tapped1.getPayload())).isEqualTo("kafkaBinderTestCommonsDelegate");
-			assertThat(new String((byte[]) tapped2.getPayload())).isEqualTo("kafkaBinderTestCommonsDelegate");
+			assertThat(new String((byte[]) tapped1.getPayload())).isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode");
+			assertThat(new String((byte[]) tapped2.getPayload())).isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode");
 		}
 		// delete one tap stream is deleted
 		input3Binding.unbind();

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

@@ -0,0 +1,82 @@
+/*
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.junit.Test;
+
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation;
+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.provisioning.KafkaTopicProvisioner;
+import org.springframework.integration.test.util.TestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * @author Gary Russell
+ * @since 1.2.2
+ *
+ */
+public class KafkaBinderUnitTests {
+
+	@Test
+	public void testPropertyOverrides() throws Exception {
+		KafkaBinderConfigurationProperties binderConfigurationProperties = new KafkaBinderConfigurationProperties();
+		AdminUtilsOperation adminUtilsOperation = mock(AdminUtilsOperation.class);
+		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties,
+				adminUtilsOperation);
+		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfigurationProperties,
+				provisioningProvider);
+		KafkaConsumerProperties consumerProps = new KafkaConsumerProperties();
+		ExtendedConsumerProperties<KafkaConsumerProperties> ecp =
+				new ExtendedConsumerProperties<KafkaConsumerProperties>(consumerProps);
+		Method method = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class,
+				String.class, ExtendedConsumerProperties.class);
+		method.setAccessible(true);
+
+		// test default for anon
+		Object factory = method.invoke(binder, true, "foo", ecp);
+		Map<?, ?> configs = TestUtils.getPropertyValue(factory, "configs", Map.class);
+		assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("latest");
+
+		// test default for named
+		factory = method.invoke(binder, false, "foo", ecp);
+		configs = TestUtils.getPropertyValue(factory, "configs", Map.class);
+		assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest");
+
+		// binder level setting
+		binderConfigurationProperties.setConfiguration(
+				Collections.<String, Object>singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"));
+		factory = method.invoke(binder, false, "foo", ecp);
+		configs = TestUtils.getPropertyValue(factory, "configs", Map.class);
+		assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("latest");
+
+		// consumer level setting
+		consumerProps.setConfiguration(Collections.singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"));
+		factory = method.invoke(binder, false, "foo", ecp);
+		configs = TestUtils.getPropertyValue(factory, "configs", Map.class);
+		assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest");
+	}
+
+}

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -5,7 +5,7 @@
  * 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
+ *      https://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,

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

@@ -0,0 +1,47 @@
+/*
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka.bootstrap;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KafkaBinderBootstrapTest {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10);
+
+	@Test
+	public void testKafkaBinderConfiguration() throws Exception {
+		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class)
+				.web(false)
+				.run("--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+					"--spring.cloud.stream.kafka.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
+		applicationContext.close();
+	}
+
+	@SpringBootApplication
+	static class SimpleApplication {
+	}
+
+}

spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties

@@ -0,0 +1,10 @@
+spring.kafka.producer.keySerializer=org.apache.kafka.common.serialization.LongSerializer
+spring.kafka.producer.valueSerializer=org.apache.kafka.common.serialization.LongSerializer
+spring.kafka.consumer.keyDeserializer=org.apache.kafka.common.serialization.LongDeserializer
+spring.kafka.consumer.valueDeserializer=org.apache.kafka.common.serialization.LongDeserializer
+spring.kafka.producer.batchSize=10
+spring.kafka.bootstrapServers=10.98.09.199:9092,10.98.09.196:9092
+spring.kafka.producer.compressionType=snappy
+# Test consumer properties
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.group-id=groupIdFromBootConfig

spring-cloud-stream-binder-kafka/src/test/resources/binder-config.properties

@@ -0,0 +1 @@
+spring.cloud.stream.kafka.binder.brokers=10.98.09.199:9082