update scst version

With a transactional binder, the producer factory should not be destroyed.

Resolves #626

.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,21 +1,21 @@
 <?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>2.0.0.RELEASE</version>
+	<version>2.0.3.RELEASE</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>2.0.0.RELEASE</version>
+		<version>2.0.6.RELEASE</version>
 		<relativePath />
 	</parent>
 	<properties>
 		<java.version>1.8</java.version>
-		<spring-kafka.version>2.1.5.RELEASE</spring-kafka.version>
+		<spring-kafka.version>2.1.10.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>3.0.3.RELEASE</spring-integration-kafka.version>
-		<kafka.version>1.0.1</kafka.version>
-		<spring-cloud-stream.version>2.0.0.RELEASE</spring-cloud-stream.version>
+		<kafka.version>1.0.2</kafka.version>
+		<spring-cloud-stream.version>2.0.3.RELEASE</spring-cloud-stream.version>
 	</properties>
 	<modules>
 		<module>spring-cloud-stream-binder-kafka</module>
@@ -155,7 +155,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>
@@ -166,7 +166,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>
@@ -174,7 +174,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>
@@ -184,7 +184,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>
@@ -195,7 +195,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>
@@ -203,7 +203,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>2.0.0.RELEASE</version>
+		<version>2.0.3.RELEASE</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-core/pom.xml

@@ -1,18 +1,18 @@
 <?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">
+	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>2.0.0.RELEASE</version>
+		<version>2.0.3.RELEASE</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>
 
 	<dependencies>

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/KafkaAdminProperties.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

@@ -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/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

@@ -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/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

@@ -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/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,

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-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.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>2.0.0.RELEASE</version>
+        <version>2.0.3.RELEASE</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/index.adoc

@@ -11,13 +11,13 @@ 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
 // ======================================================================================
 

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

@@ -17,7 +17,7 @@ Spring Cloud Stream's Apache Kafka support also includes a binder implementation
 Streams binding. With this native integration, a Spring Cloud Stream "processor" application can directly use the 
 https://kafka.apache.org/documentation/streams/developer-guide[Apache Kafka Streams] APIs in the core business logic.
 
-Kafka Streams binder implementation builds on the foundation provided by the http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafka Streams in Spring Kafka] 
+Kafka Streams binder implementation builds on the foundation provided by the https://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafka Streams in Spring Kafka] 
 project.
 
 As part of this native integration, the high-level https://docs.confluent.io/current/streams/developer-guide/dsl-api.html[Streams DSL] 
@@ -81,7 +81,7 @@ For common configuration options and properties pertaining to binder, refer to t
 
 === Kafka Streams Properties
 
-The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.binder.` 
+The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` 
 literal.
 
 configuration::
@@ -595,4 +595,18 @@ Once you gain access to this bean, then you can query for the particular state-s
 ----
 ReadOnlyKeyValueStore<Object, Object> keyValueStore =
 						queryableStoreRegistry.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
+----
+
+== Accessing the underlying KafkaStreams object
+
+`StreamBuilderFactoryBean` from spring-kafka that is responsible for constructing the `KafkaStreams` object can be accessed programmatically.
+Each `StreamBuilderFactoryBean` is registered as `stream-builder` and appended with the `StreamListener` method name.
+If your `StreamListener` method is named as `process` for example, the stream builder bean is named as `stream-builder-process`.
+Since this is a factory bean, it should be accessed by prepending an ampersand (`&`) when accessing it programmatically.
+Following is an example and it assumes the `StreamListener` method is named as `process`
+
+[source]
+----
+StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
+			KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
 ----

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

@@ -325,7 +325,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, to set `security.protocol` to `SASL_SSL`, set the following property:
@@ -337,7 +337,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 by using a JAAS configuration file and using Spring Boot properties.
 
@@ -454,7 +454,7 @@ public class Application {
 == Error Channels
 
 Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel.
-See <<binder-error-channels>> for more information.
+See <<spring-cloud-stream-overview-error-handling>> for more information.
 
 The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureException` with properties:
 
@@ -469,6 +469,6 @@ You can consume these exceptions with your own Spring Integration flow.
 
 Kafka binder module exposes the following metrics:
 
-`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag`: This metric indicates how many messages have not been yet consumed from a given binder's topic by a given consumer group.
-For example, if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, the consumer group named `myGroup` has `1000` messages waiting to be consumed from the topic calle `myTopic`.
+`spring.cloud.stream.binder.kafka.offset`: This metric indicates how many messages have not been yet consumed from a given binder's topic by a given consumer group.
+The metrics provided are based on the Mircometer metrics library. The metric contains the consumer group information, topic and the actual lag in committed offset from the latest offset on the topic.
 This metric is particularly useful for providing auto-scaling feedback to a PaaS platform.

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-streams/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-streams</artifactId>
@@ -10,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>2.0.0.RELEASE</version>
+		<version>2.0.3.RELEASE</version>
 	</parent>
 
 	<dependencies>
@@ -62,5 +62,10 @@
 			<artifactId>spring-cloud-stream-binder-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-autoconfigure-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
 	</dependencies>
-</project>
+</project>

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.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,
@@ -23,10 +23,12 @@ import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProv
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 
 /**
  * @author Soby Chacko
  */
+@Configuration
 public class KTableBinderConfiguration {
 
 	@Autowired

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.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,
@@ -73,7 +73,7 @@ class KafkaStreamsMessageConversionDelegate {
 		String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget);
 		MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
 
-		return outboundBindTarget.map((k, v) -> {
+		return outboundBindTarget.mapValues((v) -> {
 			Message<?> message = v instanceof Message<?> ? (Message<?>) v :
 					MessageBuilder.withPayload(v).build();
 			Map<String, Object> headers = new HashMap<>(message.getHeaders());
@@ -81,9 +81,9 @@ class KafkaStreamsMessageConversionDelegate {
 				headers.put(MessageHeaders.CONTENT_TYPE, contentType);
 			}
 			MessageHeaders messageHeaders = new MessageHeaders(headers);
-			return new KeyValue<>(k,
+			return 
 					messageConverter.toMessage(message.getPayload(),
-							messageHeaders).getPayload());
+							messageHeaders).getPayload();
 		});
 	}
 
@@ -137,10 +137,10 @@ class KafkaStreamsMessageConversionDelegate {
 		processErrorFromDeserialization(bindingTarget, branch[1]);
 
 		//first branch above is the branch where the messages are converted, let it go through further processing.
-		return branch[0].map((o, o2) -> {
-			KeyValue<Object, Object> objectObjectKeyValue = keyValueThreadLocal.get();
+		return branch[0].mapValues((o2) -> {
+			Object objectValue = keyValueThreadLocal.get().value;
 			keyValueThreadLocal.remove();
-			return objectObjectKeyValue;
+			return objectValue;
 		});
 	}
 

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.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,14 +20,12 @@ import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.UUID;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.Consumed;
-import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
@@ -300,18 +298,18 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
 		else {
 			LOG.info("Native decoding is disabled for " + inboundName + ". Inbound message conversion done by Spring Cloud Stream.");
 		}
-		stream = stream.map((key, value) -> {
-			KeyValue<Object, Object> keyValue;
+
+		stream = stream.mapValues(value -> {
+			Object returnValue;
 			String contentType = bindingProperties.getContentType();
 			if (!StringUtils.isEmpty(contentType) && !nativeDecoding) {
 				Message<?> message = MessageBuilder.withPayload(value)
 						.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
-				keyValue = new KeyValue<>(key, message);
+						returnValue = message;
+			} else {
+				returnValue = value;
 			}
-			else {
-				keyValue = new KeyValue<>(key, value);
-			}
-			return keyValue;
+			return returnValue;
 		});
 		return stream;
 	}
@@ -332,12 +330,11 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
 		ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory();
 		StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean();
 		streamsBuilder.setAutoStartup(false);
-		String uuid = UUID.randomUUID().toString();
 		BeanDefinition streamsBuilderBeanDefinition =
 				BeanDefinitionBuilder.genericBeanDefinition((Class<StreamsBuilderFactoryBean>) streamsBuilder.getClass(), () -> streamsBuilder)
 				.getRawBeanDefinition();
-		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + uuid, streamsBuilderBeanDefinition);
-		StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + uuid, StreamsBuilderFactoryBean.class);
+		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + method.getName(), streamsBuilderBeanDefinition);
+		StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class);
 		String group = bindingProperties.getGroup();
 		if (!StringUtils.hasText(group)) {
 			group = binderConfigurationProperties.getApplicationId();
@@ -367,7 +364,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
 		BeanDefinition streamsConfigBeanDefinition =
 				BeanDefinitionBuilder.genericBeanDefinition((Class<StreamsConfig>) streamsConfig.getClass(), () -> streamsConfig)
 						.getRawBeanDefinition();
-		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("streamsConfig-" + uuid, streamsConfigBeanDefinition);
+		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("streamsConfig-" + method.getName(), streamsConfigBeanDefinition);
 
 		streamsBuilder.setStreamsConfig(streamsConfig);
 		methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX);

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.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,7 @@ import java.util.HashSet;
 import java.util.Set;
 
 import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.errors.InvalidStateStoreException;
 import org.apache.kafka.streams.state.QueryableStoreType;
 
 /**
@@ -27,6 +28,7 @@ import org.apache.kafka.streams.state.QueryableStoreType;
  * the user applications.
  *
  * @author Soby Chacko
+ * @author Renwei Han
  * @since 2.0.0
  */
 public class QueryableStoreRegistry {
@@ -44,9 +46,14 @@ public class QueryableStoreRegistry {
 	public <T> T getQueryableStoreType(String storeName, QueryableStoreType<T> storeType) {
 
 		for (KafkaStreams kafkaStream : kafkaStreams) {
-			T store = kafkaStream.store(storeName, storeType);
-			if (store != null) {
-				return store;
+			try{
+				T store = kafkaStream.store(storeName, storeType);
+				if (store != null) {
+					return store;
+				}
+			}
+			catch (InvalidStateStoreException ignored) {
+				//pass through
 			}
 		}
 		return null;

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.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-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/DeserializationErrorHandlerByKafkaTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/DeserializtionErrorHandlerByBinderTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.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,
@@ -106,10 +106,10 @@ public class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
 		template.sendDefault("{\"id\":\"123\"}");
 		ConsumerRecord<Integer, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id");
 
-		assertThat(cr.key().equals(123));
+		assertThat(cr.key()).isEqualTo(123);
 		ObjectMapper om = new ObjectMapper();
 		Long aLong = om.readValue(cr.value(), Long.class);
-		assertThat(aLong.equals(1L));
+		assertThat(aLong).isEqualTo(1L);
 	}
 
 	@EnableBinding(KafkaStreamsProcessor.class)

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderWordCountIntegrationTests.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,
@@ -24,11 +24,14 @@ import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -48,6 +51,7 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.StreamsBuilderFactoryBean;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.handler.annotation.SendTo;
@@ -101,6 +105,12 @@ public class KafkaStreamsBinderWordCountIntegrationTests {
 				"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString());
 		try {
 			receiveAndValidate(context);
+			//Assertions on StreamBuilderFactoryBean
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
+			KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
+			ReadOnlyWindowStore<Object, Object> store = kafkaStreams.store("foo-WordCounts", QueryableStoreTypes.windowStore());
+			assertThat(store).isNotNull();
+
 		} finally {
 			context.close();
 		}

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsNativeEncodingDecodingTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkastreamsBinderPojoInputStringOutputIntegrationTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/StreamToTableJoinIntegrationTests.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-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/WordCountMultipleBranchesIntegrationTests.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/.settings/org.eclipse.jdt.ui.prefs

@@ -1,5 +1,5 @@
 eclipse.preferences.version=1
 org.eclipse.jdt.ui.ignorelowercasenames=true
-org.eclipse.jdt.ui.importorder=java;javax;com;org;org.springframework;ch.qos;\#;
+org.eclipse.jdt.ui.importorder=java;javax;com;io.micrometer;org;org.springframework;ch.qos;\#;
 org.eclipse.jdt.ui.ondemandthreshold=99
 org.eclipse.jdt.ui.staticondemandthreshold=99

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,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>2.0.0.RELEASE</version>
+		<version>2.0.3.RELEASE</version>
 	</parent>
 
 	<dependencies>
@@ -37,11 +37,6 @@
 			<artifactId>spring-boot-autoconfigure</artifactId>
 			<optional>true</optional>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-test</artifactId>
-			<scope>test</scope>
-		</dependency>
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka-clients</artifactId>

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

@@ -0,0 +1,349 @@
+/*
+ * Copyright 2017-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import org.springframework.kafka.support.AbstractKafkaHeaderMapper;
+import org.springframework.kafka.support.MimeTypeJsonDeserializer;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
+
+/**
+ * Default header mapper for Apache Kafka.
+ * Most headers in {@link org.springframework.kafka.support.KafkaHeaders} are not mapped on outbound messages.
+ * The exceptions are correlation and reply headers for request/reply
+ * messaging.
+ * Header types are added to a special header {@link #JSON_TYPES}.
+ *
+ * @author Gary Russell
+ * @since 2.0.2
+ *
+ */
+@Deprecated
+public class BinderHeaderMapper extends AbstractKafkaHeaderMapper {
+
+	private static final List<String> DEFAULT_TRUSTED_PACKAGES =
+			Arrays.asList(
+					"java.util",
+					"java.lang"
+			);
+
+	private static final List<String> DEFAULT_TO_STRING_CLASSES =
+			Arrays.asList(
+					"org.springframework.util.MimeType",
+					"org.springframework.http.MediaType"
+			);
+
+	/**
+	 * Header name for java types of other headers.
+	 */
+	public static final String JSON_TYPES = "spring_json_header_types";
+
+	private final ObjectMapper objectMapper;
+
+	private final Set<String> trustedPackages = new LinkedHashSet<>(DEFAULT_TRUSTED_PACKAGES);
+
+	private final Set<String> toStringClasses = new LinkedHashSet<>(DEFAULT_TO_STRING_CLASSES);
+
+	/**
+	 * Construct an instance with the default object mapper and default header patterns
+	 * for outbound headers; all inbound headers are mapped. The default pattern list is
+	 * {@code "!id", "!timestamp" and "*"}. In addition, most of the headers in
+	 * {@link KafkaHeaders} are never mapped as headers since they represent data in
+	 * consumer/producer records.
+	 * @see #BinderHeaderMapper(ObjectMapper)
+	 */
+	public BinderHeaderMapper() {
+		this(new ObjectMapper());
+	}
+
+	/**
+	 * Construct an instance with the provided object mapper and default header patterns
+	 * for outbound headers; all inbound headers are mapped. The patterns are applied in
+	 * order, stopping on the first match (positive or negative). Patterns are negated by
+	 * preceding them with "!". The default pattern list is
+	 * {@code "!id", "!timestamp" and "*"}. In addition, most of the headers in
+	 * {@link KafkaHeaders} are never mapped as headers since they represent data in
+	 * consumer/producer records.
+	 * @param objectMapper the object mapper.
+	 * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
+	 */
+	public BinderHeaderMapper(ObjectMapper objectMapper) {
+		this(objectMapper,
+				"!" + MessageHeaders.ID,
+				"!" + MessageHeaders.TIMESTAMP,
+				"*");
+	}
+
+	/**
+	 * Construct an instance with a default object mapper and the provided header patterns
+	 * for outbound headers; all inbound headers are mapped. The patterns are applied in
+	 * order, stopping on the first match (positive or negative). Patterns are negated by
+	 * preceding them with "!". The patterns will replace the default patterns; you
+	 * generally should not map the {@code "id" and "timestamp"} headers. Note:
+	 * most of the headers in {@link KafkaHeaders} are ever mapped as headers since they
+	 * represent data in consumer/producer records.
+	 * @param patterns the patterns.
+	 * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
+	 */
+	public BinderHeaderMapper(String... patterns) {
+		this(new ObjectMapper(), patterns);
+	}
+
+	/**
+	 * Construct an instance with the provided object mapper and the provided header
+	 * patterns for outbound headers; all inbound headers are mapped. The patterns are
+	 * applied in order, stopping on the first match (positive or negative). Patterns are
+	 * negated by preceding them with "!". The patterns will replace the default patterns;
+	 * you generally should not map the {@code "id" and "timestamp"} headers. Note: most
+	 * of the headers in {@link KafkaHeaders} are never mapped as headers since they
+	 * represent data in consumer/producer records.
+	 * @param objectMapper the object mapper.
+	 * @param patterns the patterns.
+	 * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
+	 */
+	public BinderHeaderMapper(ObjectMapper objectMapper, String... patterns) {
+		super(patterns);
+		Assert.notNull(objectMapper, "'objectMapper' must not be null");
+		Assert.noNullElements(patterns, "'patterns' must not have null elements");
+		this.objectMapper = objectMapper;
+		Module module = new SimpleModule().addDeserializer(MimeType.class, new MimeTypeJsonDeserializer(objectMapper));
+		this.objectMapper.registerModule(module);
+	}
+
+	/**
+	 * Return the object mapper.
+	 * @return the mapper.
+	 */
+	protected ObjectMapper getObjectMapper() {
+		return this.objectMapper;
+	}
+
+	/**
+	 * Provide direct access to the trusted packages set for subclasses.
+	 * @return the trusted packages.
+	 * @since 2.2
+	 */
+	protected Set<String> getTrustedPackages() {
+		return this.trustedPackages;
+	}
+
+	/**
+	 * Provide direct access to the toString() classes by subclasses.
+	 * @return the toString() classes.
+	 * @since 2.2
+	 */
+	protected Set<String> getToStringClasses() {
+		return this.toStringClasses;
+	}
+
+	/**
+	 * Add packages to the trusted packages list (default {@code java.util, java.lang}) used
+	 * when constructing objects from JSON.
+	 * If any of the supplied packages is {@code "*"}, all packages are trusted.
+	 * If a class for a non-trusted package is encountered, the header is returned to the
+	 * application with value of type {@link NonTrustedHeaderType}.
+	 * @param trustedPackages the packages to trust.
+	 */
+	public void addTrustedPackages(String... trustedPackages) {
+		if (trustedPackages != null) {
+			for (String whiteList : trustedPackages) {
+				if ("*".equals(whiteList)) {
+					this.trustedPackages.clear();
+					break;
+				}
+				else {
+					this.trustedPackages.add(whiteList);
+				}
+			}
+		}
+	}
+
+	/**
+	 * Add class names that the outbound mapper should perform toString() operations on
+	 * before mapping.
+	 * @param classNames the class names.
+	 * @since 2.2
+	 */
+	public void addToStringClasses(String... classNames) {
+		this.toStringClasses.addAll(Arrays.asList(classNames));
+	}
+
+	@Override
+	public void fromHeaders(MessageHeaders headers, Headers target) {
+		final Map<String, String> jsonHeaders = new HashMap<>();
+		headers.forEach((k, v) -> {
+			if (matches(k, v)) {
+				if (v instanceof byte[]) {
+					target.add(new RecordHeader(k, (byte[]) v));
+				}
+				else {
+					try {
+						Object value = v;
+						String className = v.getClass().getName();
+						if (this.toStringClasses.contains(className)) {
+							value = v.toString();
+							className = "java.lang.String";
+						}
+						target.add(new RecordHeader(k, getObjectMapper().writeValueAsBytes(value)));
+						jsonHeaders.put(k, className);
+					}
+					catch (Exception e) {
+						if (logger.isDebugEnabled()) {
+							logger.debug("Could not map " + k + " with type " + v.getClass().getName());
+						}
+					}
+				}
+			}
+		});
+		if (jsonHeaders.size() > 0) {
+			try {
+				target.add(new RecordHeader(JSON_TYPES, getObjectMapper().writeValueAsBytes(jsonHeaders)));
+			}
+			catch (IllegalStateException | JsonProcessingException e) {
+				logger.error("Could not add json types header", e);
+			}
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void toHeaders(Headers source, final Map<String, Object> headers) {
+		Map<String, String> types = null;
+		Iterator<Header> iterator = source.iterator();
+		while (iterator.hasNext()) {
+			Header next = iterator.next();
+			if (next.key().equals(JSON_TYPES)) {
+				try {
+					types = getObjectMapper().readValue(next.value(), HashMap.class);
+				}
+				catch (IOException e) {
+					logger.error("Could not decode json types: " + new String(next.value()), e);
+				}
+				break;
+			}
+		}
+		final Map<String, String> jsonTypes = types;
+		source.forEach(h -> {
+			if (!(h.key().equals(JSON_TYPES))) {
+				if (jsonTypes != null && jsonTypes.containsKey(h.key())) {
+					Class<?> type = Object.class;
+					String requestedType = jsonTypes.get(h.key());
+					boolean trusted = false;
+					try {
+						trusted = trusted(requestedType);
+						if (trusted) {
+							type = ClassUtils.forName(requestedType, null);
+						}
+					}
+					catch (Exception e) {
+						logger.error("Could not load class for header: " + h.key(), e);
+					}
+					if (trusted) {
+						try {
+							headers.put(h.key(), getObjectMapper().readValue(h.value(), type));
+						}
+						catch (IOException e) {
+							logger.error("Could not decode json type: " + new String(h.value()) + " for key: " + h.key(),
+									e);
+							headers.put(h.key(), h.value());
+						}
+					}
+					else {
+						headers.put(h.key(), new NonTrustedHeaderType(h.value(), requestedType));
+					}
+				}
+				else {
+					headers.put(h.key(), h.value());
+				}
+			}
+		});
+	}
+
+	protected boolean trusted(String requestedType) {
+		if (!this.trustedPackages.isEmpty()) {
+			int lastDot = requestedType.lastIndexOf(".");
+			if (lastDot < 0) {
+				return false;
+			}
+			String packageName = requestedType.substring(0, lastDot);
+			for (String trustedPackage : this.trustedPackages) {
+				if (packageName.equals(trustedPackage) || packageName.startsWith(trustedPackage + ".")) {
+					return true;
+				}
+			}
+			return false;
+		}
+		return true;
+	}
+
+	/**
+	 * Represents a header that could not be decoded due to an untrusted type.
+	 */
+	public static class NonTrustedHeaderType {
+
+		private final byte[] headerValue;
+
+		private final String untrustedType;
+
+		NonTrustedHeaderType(byte[] headerValue, String untrustedType) { // NOSONAR
+			this.headerValue = headerValue; // NOSONAR
+			this.untrustedType = untrustedType;
+		}
+
+		public byte[] getHeaderValue() {
+			return this.headerValue;
+		}
+
+		public String getUntrustedType() {
+			return this.untrustedType;
+		}
+
+		@Override
+		public String toString() {
+			try {
+				return "NonTrustedHeaderType [headerValue=" + new String(this.headerValue, StandardCharsets.UTF_8)
+						+ ", untrustedType=" + this.untrustedType + "]";
+			}
+			catch (Exception e) {
+				return "NonTrustedHeaderType [headerValue=" + Arrays.toString(this.headerValue) + ", untrustedType="
+						+ this.untrustedType + "]";
+			}
+		}
+
+	}
+
+}

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,

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      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,
@@ -78,25 +78,31 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
 			public Health call() {
 				try {
 					if (metadataConsumer == null) {
-						metadataConsumer = consumerFactory.createConsumer();
-					}
-					Set<String> downMessages = new HashSet<>();
-					for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) {
-						List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
-						for (PartitionInfo partitionInfo : partitionInfos) {
-							if (KafkaBinderHealthIndicator.this.binder.getTopicsInUse().get(topic).getPartitionInfos()
-									.contains(partitionInfo) && partitionInfo.leader().id() == -1) {
-								downMessages.add(partitionInfo.toString());
+						synchronized(KafkaBinderHealthIndicator.this) {
+							if (metadataConsumer == null) {
+								metadataConsumer = consumerFactory.createConsumer();
 							}
 						}
 					}
-					if (downMessages.isEmpty()) {
-						return Health.up().build();
-					}
-					else {
-						return Health.down()
-							.withDetail("Following partitions in use have no leaders: ", downMessages.toString())
-							.build();
+					synchronized (metadataConsumer) {
+						Set<String> downMessages = new HashSet<>();
+						for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) {
+							List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
+							for (PartitionInfo partitionInfo : partitionInfos) {
+								if (KafkaBinderHealthIndicator.this.binder.getTopicsInUse().get(topic).getPartitionInfos()
+										.contains(partitionInfo) && partitionInfo.leader().id() == -1) {
+									downMessages.add(partitionInfo.toString());
+								}
+							}
+						}
+						if (downMessages.isEmpty()) {
+							return Health.up().build();
+						}
+						else {
+							return Health.down()
+								.withDetail("Following partitions in use have no leaders: ", downMessages.toString())
+								.build();
+						}
 					}
 				}
 				catch (Exception e) {

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.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,11 +20,17 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+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 io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.TimeGauge;
 import io.micrometer.core.instrument.binder.MeterBinder;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -50,9 +56,13 @@ import org.springframework.util.ObjectUtils;
  * @author Artem Bilan
  * @author Oleg Zhurakousky
  * @author Jon Schneider
+ * @author Thomas Cheyney
+ * @author Gary Russell
  */
 public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<BindingCreatedEvent> {
 
+	private static final int DEFAULT_TIMEOUT = 60;
+
 	private final static Log LOG = LogFactory.getLog(KafkaBinderMetrics.class);
 
 	static final String METRIC_NAME = "spring.cloud.stream.binder.kafka.offset";
@@ -65,6 +75,10 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
 
 	private final MeterRegistry meterRegistry;
 
+	private Consumer<?, ?> metadataConsumer;
+
+	private int timeout = DEFAULT_TIMEOUT;
+
 	public KafkaBinderMetrics(KafkaMessageChannelBinder binder,
 			KafkaBinderConfigurationProperties binderConfigurationProperties,
 			ConsumerFactory<?, ?> defaultConsumerFactory, @Nullable MeterRegistry meterRegistry) {
@@ -81,6 +95,10 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
 		this(binder, binderConfigurationProperties, null, null);
 	}
 
+	public void setTimeout(int timeout) {
+		this.timeout = timeout;
+	}
+
 	@Override
 	public void bindTo(MeterRegistry registry) {
 		for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder.getTopicsInUse()
@@ -103,30 +121,56 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
 	}
 
 	private double calculateConsumerLagOnTopic(String topic, String group) {
-		long lag = 0;
-		try (Consumer<?, ?> metadataConsumer = createConsumerFactory(group).createConsumer()) {
-			List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
-			List<TopicPartition> topicPartitions = new LinkedList<>();
-			for (PartitionInfo partitionInfo : partitionInfos) {
-				topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
-			}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		Future<Long> future = exec.submit(() -> {
 
-			Map<TopicPartition, Long> endOffsets = metadataConsumer.endOffsets(topicPartitions);
-
-			for (Map.Entry<TopicPartition, Long> endOffset : endOffsets.entrySet()) {
-				OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey());
-				if (current != null) {
-					lag += endOffset.getValue() - current.offset();
+			long lag = 0;
+			try {
+				if (metadataConsumer == null) {
+					synchronized(KafkaBinderMetrics.this) {
+						if (metadataConsumer == null) {
+							metadataConsumer = createConsumerFactory(group).createConsumer();
+						}
+					}
 				}
-				else {
-					lag += endOffset.getValue();
+				synchronized (metadataConsumer) {
+					List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
+					List<TopicPartition> topicPartitions = new LinkedList<>();
+					for (PartitionInfo partitionInfo : partitionInfos) {
+						topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
+					}
+
+					Map<TopicPartition, Long> endOffsets = metadataConsumer.endOffsets(topicPartitions);
+
+					for (Map.Entry<TopicPartition, Long> endOffset : endOffsets.entrySet()) {
+						OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey());
+						if (current != null) {
+							lag += endOffset.getValue() - current.offset();
+						}
+						else {
+							lag += endOffset.getValue();
+						}
+					}
 				}
 			}
+			catch (Exception e) {
+				LOG.debug("Cannot generate metric for topic: " + topic, e);
+			}
+			return lag;
+		});
+		try {
+			return future.get(this.timeout, TimeUnit.SECONDS);
 		}
-		catch (Exception e) {
-			LOG.debug("Cannot generate metric for topic: " + topic, e);
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return 0L;
+		}
+		catch (ExecutionException | TimeoutException e) {
+			return 0L;
+		}
+		finally {
+			exec.shutdownNow();
 		}
-		return lag;
 	}
 
 	private ConsumerFactory<?, ?> createConsumerFactory(String group) {

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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,
@@ -90,7 +90,6 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMo
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
 import org.springframework.kafka.listener.config.ContainerProperties;
-import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.ProducerListener;
@@ -215,7 +214,9 @@ public class KafkaMessageChannelBinder extends
 					Producer<byte[], byte[]> producer = producerFB.createProducer();
 					List<PartitionInfo> partitionsFor = producer.partitionsFor(destination.getName());
 					producer.close();
-					((DisposableBean) producerFB).destroy();
+					if (this.transactionManager == null) {
+						((DisposableBean) producerFB).destroy();
+					}
 					return partitionsFor;
 				});
 		this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions));
@@ -268,10 +269,10 @@ public class KafkaMessageChannelBinder extends
 				if (!patterns.contains("!" + MessageHeaders.ID)) {
 					patterns.add(0, "!" + MessageHeaders.ID);
 				}
-				mapper = new DefaultKafkaHeaderMapper(patterns.toArray(new String[patterns.size()]));
+				mapper = new BinderHeaderMapper(patterns.toArray(new String[patterns.size()]));
 			}
 			else {
-				mapper = new DefaultKafkaHeaderMapper();
+				mapper = new BinderHeaderMapper();
 			}
 		}
 		handler.setHeaderMapper(mapper);
@@ -471,6 +472,7 @@ public class KafkaMessageChannelBinder extends
 	@Override
 	protected PolledConsumerResources createPolledConsumerResources(String name, String group,
 			ConsumerDestination destination, ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
+
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(!anonymous || !consumerProperties.getExtension().isEnableDlq(),
 				"DLQ support is not available for anonymous subscriptions");
@@ -480,6 +482,11 @@ public class KafkaMessageChannelBinder extends
 		KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory, destination.getName());
 		source.setMessageConverter(getMessageConverter(consumerProperties));
 		source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq());
+		String clientId = name;
+		if (consumerProperties.getExtension().getConfiguration().containsKey(ConsumerConfig.CLIENT_ID_CONFIG)) {
+			clientId = consumerProperties.getExtension().getConfiguration().get(ConsumerConfig.CLIENT_ID_CONFIG);
+		}
+		source.setClientId(clientId);
 
 		// I copied this from the regular consumer - it looks bogus to me - includes all partitions
 		// not just the ones this binding is listening to; doesn't seem right for a health check.
@@ -547,7 +554,7 @@ public class KafkaMessageChannelBinder extends
 					KafkaHeaderMapper.class);
 		}
 		if (mapper == null) {
-			DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper() {
+			BinderHeaderMapper headerMapper = new BinderHeaderMapper() {
 
 				@Override
 				public void toHeaders(Headers source, Map<String, Object> headers) {

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,
@@ -18,6 +18,8 @@ package org.springframework.cloud.stream.binder.kafka.config;
 
 import java.io.IOException;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.binder.MeterBinder;
 
@@ -56,7 +58,8 @@ import org.springframework.kafka.support.ProducerListener;
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
-@Import({KafkaAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, KafkaBinderHealthIndicatorConfiguration.class })
+@Import({ KafkaAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class,
+		KafkaBinderHealthIndicatorConfiguration.class })
 @EnableConfigurationProperties({ KafkaExtendedBindingProperties.class })
 public class KafkaBinderConfiguration {
 
@@ -81,7 +84,7 @@ public class KafkaBinderConfiguration {
 
 	@Bean
 	KafkaMessageChannelBinder kafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
-			KafkaTopicProvisioner provisioningProvider) {
+														KafkaTopicProvisioner provisioningProvider) {
 
 		KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
 				configurationProperties, provisioningProvider);
@@ -97,8 +100,34 @@ public class KafkaBinderConfiguration {
 	}
 
 	@Bean
-	public KafkaJaasLoginModuleInitializer jaasInitializer() throws IOException {
-		return new KafkaJaasLoginModuleInitializer();
+	@ConditionalOnMissingBean(KafkaJaasLoginModuleInitializer.class)
+	public KafkaJaasLoginModuleInitializer jaasInitializer(KafkaBinderConfigurationProperties configurationProperties) throws IOException {
+		KafkaJaasLoginModuleInitializer kafkaJaasLoginModuleInitializer = new KafkaJaasLoginModuleInitializer();
+		JaasLoginModuleConfiguration jaas = configurationProperties.getJaas();
+		if (jaas != null) {
+			kafkaJaasLoginModuleInitializer.setLoginModule(jaas.getLoginModule());
+
+			KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = null;
+			AppConfigurationEntry.LoginModuleControlFlag controlFlagValue = jaas.getControlFlagValue();
+
+			if (AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL.equals(controlFlagValue)) {
+				controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.OPTIONAL;
+			}
+			else if (AppConfigurationEntry.LoginModuleControlFlag.REQUIRED.equals(controlFlagValue)) {
+				controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUIRED;
+			}
+			else if (AppConfigurationEntry.LoginModuleControlFlag.REQUISITE.equals(controlFlagValue)) {
+				controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUISITE;
+			}
+			else if (AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT.equals(controlFlagValue)) {
+				controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.SUFFICIENT;
+			}
+			if (controlFlag != null) {
+				kafkaJaasLoginModuleInitializer.setControlFlag(controlFlag);
+			}
+			kafkaJaasLoginModuleInitializer.setOptions(jaas.getOptions());
+		}
+		return kafkaJaasLoginModuleInitializer;
 	}
 
 	/**

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.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/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/AdminConfigTests.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

@@ -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,
@@ -41,7 +41,9 @@ import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -74,11 +76,11 @@ public class KafkaBinderAutoConfigurationPropertiesTest {
 		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("key.deserializer") == null);
-		assertTrue(producerConfigs.get("value.serializer").equals(LongSerializer.class));
-		assertTrue(producerConfigs.get("value.deserializer") == null);
-		assertTrue(producerConfigs.get("compression.type").equals("snappy"));
+		assertEquals(producerConfigs.get("key.serializer"), LongSerializer.class);
+		assertNull(producerConfigs.get("key.deserializer"));
+		assertEquals(producerConfigs.get("value.serializer"), LongSerializer.class);
+		assertNull(producerConfigs.get("value.deserializer"));
+		assertEquals("snappy", producerConfigs.get("compression.type"));
 		List<String> bootstrapServers = new ArrayList<>();
 		bootstrapServers.add("10.98.09.199:9092");
 		bootstrapServers.add("10.98.09.196:9092");
@@ -95,12 +97,12 @@ public class KafkaBinderAutoConfigurationPropertiesTest {
 		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("key.serializer") == null);
-		assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class));
-		assertTrue(consumerConfigs.get("value.serialized") == null);
-		assertTrue(consumerConfigs.get("group.id").equals("groupIdFromBootConfig"));
-		assertTrue(consumerConfigs.get("auto.offset.reset").equals("earliest"));
+		assertEquals(consumerConfigs.get("key.deserializer"), LongDeserializer.class);
+		assertNull(consumerConfigs.get("key.serializer"));
+		assertEquals(consumerConfigs.get("value.deserializer"), LongDeserializer.class);
+		assertNull(consumerConfigs.get("value.serialized"));
+		assertEquals("groupIdFromBootConfig", consumerConfigs.get("group.id"));
+		assertEquals("earliest", consumerConfigs.get("auto.offset.reset"));
 		assertTrue((((List<String>) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers)));
 	}
 

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.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,
@@ -41,6 +41,7 @@ import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
@@ -75,11 +76,11 @@ public class KafkaBinderConfigurationPropertiesTest {
 		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"));
+		assertEquals("12345", producerConfigs.get("batch.size"));;
+		assertEquals("100", producerConfigs.get("linger.ms"));
+		assertEquals(producerConfigs.get("key.serializer"), ByteArraySerializer.class);
+		assertEquals(producerConfigs.get("value.serializer"), ByteArraySerializer.class);
+		assertEquals("gzip", producerConfigs.get("compression.type"));
 		List<String> bootstrapServers = new ArrayList<>();
 		bootstrapServers.add("10.98.09.199:9082");
 		assertTrue((((String) producerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082")));
@@ -95,8 +96,8 @@ public class KafkaBinderConfigurationPropertiesTest {
 		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));
+		assertEquals(consumerConfigs.get("key.deserializer"), ByteArrayDeserializer.class);
+		assertEquals(consumerConfigs.get("value.deserializer"), 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

@@ -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/KafkaBinderHealthIndicatorTest.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/KafkaBinderMetricsTest.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,
@@ -23,9 +23,11 @@ import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.TimeGauge;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
@@ -33,6 +35,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
 import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation;
@@ -43,6 +46,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Henryk Konsek
+ * @author Thomas Cheyney
  */
 public class KafkaBinderMetricsTest {
 
@@ -123,6 +127,37 @@ public class KafkaBinderMetricsTest {
 		assertThat(meterRegistry.getMeters()).isEmpty();
 	}
 
+	@Test
+	public void createsConsumerOnceWhenInvokedMultipleTimes() {
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
+
+		metrics.bindTo(meterRegistry);
+
+		TimeGauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME).tag("group", "group").tag("topic", TEST_TOPIC).timeGauge();
+		gauge.value(TimeUnit.MILLISECONDS);
+		assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(1000.0);
+
+		org.mockito.Mockito.verify(this.consumerFactory).createConsumer();
+	}
+
+	@Test
+	public void consumerCreationFailsFirstTime() {
+		org.mockito.BDDMockito.given(consumerFactory.createConsumer()).willThrow(KafkaException.class)
+				.willReturn(consumer);
+
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
+
+		metrics.bindTo(meterRegistry);
+
+		TimeGauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME).tag("group", "group").tag("topic", TEST_TOPIC).timeGauge();
+		assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(0);
+		assertThat(gauge.value(TimeUnit.MILLISECONDS)).isEqualTo(1000.0);
+
+		org.mockito.Mockito.verify(this.consumerFactory, Mockito.times(2)).createConsumer();
+	}
+
 	private List<PartitionInfo> partitions(Node... nodes) {
 		List<PartitionInfo> partitions = new ArrayList<>();
 		for (int i = 0; i < nodes.length; i++) {

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

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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,
@@ -122,7 +122,6 @@ import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
-import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
@@ -317,10 +316,10 @@ public class KafkaBinderTests extends
 
 		moduleOutputChannel.send(message);
 		CountDownLatch latch = new CountDownLatch(1);
-		AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
+		AtomicReference<Message<String>> inboundMessageRef = new AtomicReference<>();
 		moduleInputChannel.subscribe(message1 -> {
 			try {
-				inboundMessageRef.set((Message<byte[]>) message1);
+				inboundMessageRef.set((Message<String>) message1);
 			}
 			finally {
 				latch.countDown();
@@ -330,13 +329,13 @@ public class KafkaBinderTests extends
 
 
 		Assertions.assertThat(inboundMessageRef.get()).isNotNull();
-		Assertions.assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo");
+		Assertions.assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo");
 		Assertions.assertThat(inboundMessageRef.get().getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
 		Assertions.assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE))
 				.isEqualTo(MimeTypeUtils.TEXT_PLAIN);
-		Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")).isInstanceOf(MimeType.class);
-		MimeType actual = (MimeType) inboundMessageRef.get().getHeaders().get("foo");
-		Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
+		Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")).isInstanceOf(String.class);
+		String actual = (String) inboundMessageRef.get().getHeaders().get("foo");
+		Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN.toString());
 		producerBinding.unbind();
 		consumerBinding.unbind();
 	}
@@ -368,10 +367,10 @@ public class KafkaBinderTests extends
 				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
 		moduleOutputChannel.send(message);
 		CountDownLatch latch = new CountDownLatch(1);
-		AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
+		AtomicReference<Message<String>> inboundMessageRef = new AtomicReference<>();
 		moduleInputChannel.subscribe(message1 -> {
 			try {
-				inboundMessageRef.set((Message<byte[]>) message1);
+				inboundMessageRef.set((Message<String>) message1);
 			}
 			finally {
 				latch.countDown();
@@ -380,7 +379,7 @@ public class KafkaBinderTests extends
 		Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
 
 		assertThat(inboundMessageRef.get()).isNotNull();
-		assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo");
+		assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo");
 		assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE))
 				.isEqualTo(MimeTypeUtils.TEXT_PLAIN);
 		producerBinding.unbind();
@@ -426,6 +425,13 @@ public class KafkaBinderTests extends
 		assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo");
 		assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE))
 				.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
+
+		Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = ((KafkaTestBinder)binder).getCoreBinder().getTopicsInUse();
+		assertThat(topicsInUse.keySet()).contains("foo.bar");
+		KafkaMessageChannelBinder.TopicInformation topic = topicsInUse.get("foo.bar");
+		assertThat(topic.isConsumerTopic()).isTrue();
+		assertThat(topic.getConsumerGroup()).isEqualTo("testSendAndReceive");
+
 		producerBinding.unbind();
 		consumerBinding.unbind();
 	}
@@ -2330,10 +2336,10 @@ public class KafkaBinderTests extends
 			binderBindUnbindLatency();
 			moduleOutputChannel.send(message);
 			CountDownLatch latch = new CountDownLatch(1);
-			AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
+			AtomicReference<Message<String>> inboundMessageRef = new AtomicReference<>();
 			moduleInputChannel.subscribe(message1 -> {
 				try {
-					inboundMessageRef.set((Message<byte[]>) message1);
+					inboundMessageRef.set((Message<String>) message1);
 				}
 				finally {
 					latch.countDown();
@@ -2342,7 +2348,7 @@ public class KafkaBinderTests extends
 			Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
 
 			assertThat(inboundMessageRef.get()).isNotNull();
-			assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("test");
+			assertThat(inboundMessageRef.get().getPayload()).isEqualTo("test");
 			assertThat(inboundMessageRef.get().getHeaders()).containsEntry("contentType", MimeTypeUtils.TEXT_PLAIN);
 		}
 		finally {
@@ -2454,8 +2460,9 @@ public class KafkaBinderTests extends
 	public void testPolledConsumer() throws Exception {
 		KafkaTestBinder binder = getBinder();
 		PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(this.messageConverter);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProps = createConsumerProperties();
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollable", "group",
-				inboundBindTarget, createConsumerProperties());
+				inboundBindTarget, consumerProps);
 		Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
 		KafkaTemplate template = new KafkaTemplate(new DefaultKafkaProducerFactory<>(producerProps));
 		template.send("pollable", "testPollable");
@@ -2470,7 +2477,14 @@ public class KafkaBinderTests extends
 			Thread.sleep(100);
 		}
 		assertThat(polled).isTrue();
+		// Bind a second pollable consumer GH-521
+		consumerProps.getExtension().getConfiguration().put(ConsumerConfig.CLIENT_ID_CONFIG, "pollable2");
+		PollableSource<MessageHandler> second = new DefaultPollableMessageSource(this.messageConverter);
+		Binding<PollableSource<MessageHandler>> binding2 = binder.bindPollableConsumer("pollable2",
+				"group-polledConsumer2", second, consumerProps);
+		second.poll(m -> { });
 		binding.unbind();
+		binding2.unbind();
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.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/KafkaTestBinder.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,