Release 1.1.1.RELEASE

Signed-off-by: Marius Bogoevici <mbogoevici@pivotal.io>
Update mvnw
2017-01-11 22:32:42 -05:00 · 2017-01-11 22:13:52 -05:00 · 2017-01-11 22:13:29 -05:00 · 2017-01-11 22:07:38 -05:00 · 2017-01-04 22:33:38 -05:00 · 2016-12-16 18:13:33 -05:00
35 changed files with 2513 additions and 833 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -18,6 +18,7 @@ _site/
 *.ipr
 *.iws
 .idea/*
+*/.idea
 .factorypath
 dump.rdb
 .apt_generated
--- a/README.adoc
+++ b/README.adoc
@@ -1,3 +1 @@
-# spring-cloud-stream-binder-kafka
-
-Spring Cloud Stream Binder implementation for Kafka
+Spring Cloud Stream Binder for Apache Kafka
--- a/25
+++ b/25
@@ -57,27 +57,27 @@ case "`uname`" in
           #
           # Look for the Apple JDKs first to preserve the existing behaviour, and then look
           # for the new JDKs provided by Oracle.
-           # 
+           #
           if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK ] ; then
             #
             # Apple JDKs
             #
             export JAVA_HOME=/System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK/Home
           fi
-           
+
           if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Java/JavaVirtualMachines/CurrentJDK ] ; then
             #
             # Apple JDKs
             #
             export JAVA_HOME=/System/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
           fi
-             
+
           if [ -z "$JAVA_HOME" ] && [ -L "/Library/Java/JavaVirtualMachines/CurrentJDK" ] ; then
             #
             # Oracle JDKs
             #
             export JAVA_HOME=/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
-           fi           
+           fi

           if [ -z "$JAVA_HOME" ] && [ -x "/usr/libexec/java_home" ]; then
             #
@@ -219,16 +219,27 @@ concat_lines() {
 export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-$(find_maven_basedir)}
 MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"

-# Provide a "standardized" way to retrieve the CLI args that will 
+# Provide a "standardized" way to retrieve the CLI args that will
 # work with both Windows and non-Windows executions.
 MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
 export MAVEN_CMD_LINE_ARGS

 WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain

+echo "Running version check"
+VERSION=$( sed '\!<parent!,\!</parent!d' `dirname $0`/pom.xml | grep '<version' | head -1 | sed -e 's/.*<version>//' -e 's!</version>.*$!!' )
+echo "The found version is [${VERSION}]"
+
+if echo $VERSION | egrep -q 'M|RC'; then
+    echo Activating \"milestone\" profile for version=\"$VERSION\"
+    echo $MAVEN_ARGS | grep -q milestone || MAVEN_ARGS="$MAVEN_ARGS -Pmilestone"
+else
+    echo Deactivating \"milestone\" profile for version=\"$VERSION\"
+    echo $MAVEN_ARGS | grep -q milestone && MAVEN_ARGS=$(echo $MAVEN_ARGS | sed -e 's/-Pmilestone//')
+fi
+
 exec "$JAVACMD" \
  $MAVEN_OPTS \
  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
  "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
-  ${WRAPPER_LAUNCHER} "$@"
-
+  ${WRAPPER_LAUNCHER} ${MAVEN_ARGS} "$@"
--- a/mvnw.cmd
+++ b/mvnw.cmd
@@ -1,234 +1,145 @@
-#!/bin/sh
-# ----------------------------------------------------------------------------
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# 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
-#
-# 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.
-# ----------------------------------------------------------------------------
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@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
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------

-# ----------------------------------------------------------------------------
-# Maven2 Start Up Batch script
-#
-# Required ENV vars:
-# ------------------
-#   JAVA_HOME - location of a JDK home dir
-#
-# Optional ENV vars
-# -----------------
-#   M2_HOME - location of maven2's installed home dir
-#   MAVEN_OPTS - parameters passed to the Java VM when running Maven
-#     e.g. to debug Maven itself, use
-#       set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
-#   MAVEN_SKIP_RC - flag to disable loading of mavenrc files
-# ----------------------------------------------------------------------------
+@REM ----------------------------------------------------------------------------
+@REM Maven2 Start Up Batch script
+@REM
+@REM Required ENV vars:
+@REM JAVA_HOME - location of a JDK home dir
+@REM
+@REM Optional ENV vars
+@REM M2_HOME - location of maven2's installed home dir
+@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
+@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a key stroke before ending
+@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
+@REM     e.g. to debug Maven itself, use
+@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+@REM ----------------------------------------------------------------------------

-if [ -z "$MAVEN_SKIP_RC" ] ; then
+@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
+@echo off
+@REM enable echoing my setting MAVEN_BATCH_ECHO to 'on'
+@if "%MAVEN_BATCH_ECHO%" == "on"  echo %MAVEN_BATCH_ECHO%

-  if [ -f /etc/mavenrc ] ; then
-    . /etc/mavenrc
-  fi
+@REM set %HOME% to equivalent of $HOME
+if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")

-  if [ -f "$HOME/.mavenrc" ] ; then
-    . "$HOME/.mavenrc"
-  fi
+@REM Execute a user defined script before this one
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
+@REM check for pre script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat"
+if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd"
+:skipRcPre

-fi
+@setlocal

-# OS specific support.  $var _must_ be set to either true or false.
-cygwin=false;
-darwin=false;
-mingw=false
-case "`uname`" in
-  CYGWIN*) cygwin=true ;;
-  MINGW*) mingw=true;;
-  Darwin*) darwin=true
-           #
-           # Look for the Apple JDKs first to preserve the existing behaviour, and then look
-           # for the new JDKs provided by Oracle.
-           # 
-           if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK ] ; then
-             #
-             # Apple JDKs
-             #
-             export JAVA_HOME=/System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK/Home
-           fi
-           
-           if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Java/JavaVirtualMachines/CurrentJDK ] ; then
-             #
-             # Apple JDKs
-             #
-             export JAVA_HOME=/System/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
-           fi
-             
-           if [ -z "$JAVA_HOME" ] && [ -L "/Library/Java/JavaVirtualMachines/CurrentJDK" ] ; then
-             #
-             # Oracle JDKs
-             #
-             export JAVA_HOME=/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
-           fi           
+set ERROR_CODE=0

-           if [ -z "$JAVA_HOME" ] && [ -x "/usr/libexec/java_home" ]; then
-             #
-             # Apple JDKs
-             #
-             export JAVA_HOME=`/usr/libexec/java_home`
-           fi
-           ;;
-esac
+@REM To isolate internal variables from possible post scripts, we use another setlocal
+@setlocal

-if [ -z "$JAVA_HOME" ] ; then
-  if [ -r /etc/gentoo-release ] ; then
-    JAVA_HOME=`java-config --jre-home`
-  fi
-fi
+@REM ==== START VALIDATION ====
+if not "%JAVA_HOME%" == "" goto OkJHome

-if [ -z "$M2_HOME" ] ; then
-  ## resolve links - $0 may be a link to maven's home
-  PRG="$0"
+echo.
+echo Error: JAVA_HOME not found in your environment. >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error

-  # need this for relative symlinks
-  while [ -h "$PRG" ] ; do
-    ls=`ls -ld "$PRG"`
-    link=`expr "$ls" : '.*-> \(.*\)$'`
-    if expr "$link" : '/.*' > /dev/null; then
-      PRG="$link"
-    else
-      PRG="`dirname "$PRG"`/$link"
-    fi
-  done
+:OkJHome
+if exist "%JAVA_HOME%\bin\java.exe" goto init

-  saveddir=`pwd`
+echo.
+echo Error: JAVA_HOME is set to an invalid directory. >&2
+echo JAVA_HOME = "%JAVA_HOME%" >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error

-  M2_HOME=`dirname "$PRG"`/..
+@REM ==== END VALIDATION ====

-  # make it fully qualified
-  M2_HOME=`cd "$M2_HOME" && pwd`
+:init

-  cd "$saveddir"
-  # echo Using m2 at $M2_HOME
-fi
+set MAVEN_CMD_LINE_ARGS=%*

-# For Cygwin, ensure paths are in UNIX format before anything is touched
-if $cygwin ; then
-  [ -n "$M2_HOME" ] &&
-    M2_HOME=`cygpath --unix "$M2_HOME"`
-  [ -n "$JAVA_HOME" ] &&
-    JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
-  [ -n "$CLASSPATH" ] &&
-    CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
-fi
+@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
+@REM Fallback to current working directory if not found.

-# For Migwn, ensure paths are in UNIX format before anything is touched
-if $mingw ; then
-  [ -n "$M2_HOME" ] &&
-    M2_HOME="`(cd "$M2_HOME"; pwd)`"
-  [ -n "$JAVA_HOME" ] &&
-    JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
-  # TODO classpath?
-fi
+set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
+IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir

-if [ -z "$JAVA_HOME" ]; then
-  javaExecutable="`which javac`"
-  if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
-    # readlink(1) is not available as standard on Solaris 10.
-    readLink=`which readlink`
-    if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
-      if $darwin ; then
-        javaHome="`dirname \"$javaExecutable\"`"
-        javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
-      else
-        javaExecutable="`readlink -f \"$javaExecutable\"`"
-      fi
-      javaHome="`dirname \"$javaExecutable\"`"
-      javaHome=`expr "$javaHome" : '\(.*\)/bin'`
-      JAVA_HOME="$javaHome"
-      export JAVA_HOME
-    fi
-  fi
-fi
+set EXEC_DIR=%CD%
+set WDIR=%EXEC_DIR%
+:findBaseDir
+IF EXIST "%WDIR%"\.mvn goto baseDirFound
+cd ..
+IF "%WDIR%"=="%CD%" goto baseDirNotFound
+set WDIR=%CD%
+goto findBaseDir

-if [ -z "$JAVACMD" ] ; then
-  if [ -n "$JAVA_HOME"  ] ; then
-    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
-      # IBM's JDK on AIX uses strange locations for the executables
-      JAVACMD="$JAVA_HOME/jre/sh/java"
-    else
-      JAVACMD="$JAVA_HOME/bin/java"
-    fi
-  else
-    JAVACMD="`which java`"
-  fi
-fi
+:baseDirFound
+set MAVEN_PROJECTBASEDIR=%WDIR%
+cd "%EXEC_DIR%"
+goto endDetectBaseDir

-if [ ! -x "$JAVACMD" ] ; then
-  echo "Error: JAVA_HOME is not defined correctly." >&2
-  echo "  We cannot execute $JAVACMD" >&2
-  exit 1
-fi
+:baseDirNotFound
+set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
+cd "%EXEC_DIR%"

-if [ -z "$JAVA_HOME" ] ; then
-  echo "Warning: JAVA_HOME environment variable is not set."
-fi
+:endDetectBaseDir

-CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
+IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig

-# For Cygwin, switch paths to Windows format before running java
-if $cygwin; then
-  [ -n "$M2_HOME" ] &&
-    M2_HOME=`cygpath --path --windows "$M2_HOME"`
-  [ -n "$JAVA_HOME" ] &&
-    JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
-  [ -n "$CLASSPATH" ] &&
-    CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
-fi
+@setlocal EnableExtensions EnableDelayedExpansion
+for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
+@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%

-# traverses directory structure from process work directory to filesystem root
-# first directory with .mvn subdirectory is considered project base directory
-find_maven_basedir() {
-  local basedir=$(pwd)
-  local wdir=$(pwd)
-  while [ "$wdir" != '/' ] ; do
-    if [ -d "$wdir"/.mvn ] ; then
-      basedir=$wdir
-      break
-    fi
-    wdir=$(cd "$wdir/.."; pwd)
-  done
-  echo "${basedir}"
-}
+:endReadAdditionalConfig

-# concatenates all lines of a file
-concat_lines() {
-  if [ -f "$1" ]; then
-    echo "$(tr -s '\n' ' ' < "$1")"
-  fi
-}
+SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"

-export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-$(find_maven_basedir)}
-MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
+set WRAPPER_JAR="".\.mvn\wrapper\maven-wrapper.jar""
+set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain

-# Provide a "standardized" way to retrieve the CLI args that will 
-# work with both Windows and non-Windows executions.
-MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
-export MAVEN_CMD_LINE_ARGS
+%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CMD_LINE_ARGS%
+if ERRORLEVEL 1 goto error
+goto end

-WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+:error
+set ERROR_CODE=1

-exec "$JAVACMD" \
-  $MAVEN_OPTS \
-  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
-  "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
-  ${WRAPPER_LAUNCHER} "$@"
+:end
+@endlocal & set ERROR_CODE=%ERROR_CODE%

+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost
+@REM check for post script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat"
+if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd"
+:skipRcPost
+
+@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
+if "%MAVEN_BATCH_PAUSE%" == "on" pause
+
+if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE%
+
+exit /B %ERROR_CODE%
--- a/pom.xml
+++ b/pom.xml
@@ -2,36 +2,94 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-	<version>1.1.0.M1</version>
+	<version>1.1.1.RELEASE</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
-		<artifactId>spring-cloud-stream-parent</artifactId>
-		<version>1.1.0.M1</version>
+		<artifactId>spring-cloud-build</artifactId>
+		<version>1.2.1.RELEASE</version>
 		<relativePath />
 	</parent>
 	<properties>
 		<java.version>1.7</java.version>
 		<kafka.version>0.9.0.1</kafka.version>
-		<spring-kafka.version>1.0.3.RELEASE</spring-kafka.version>
+		<spring-kafka.version>1.0.5.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>2.0.1.RELEASE</spring-integration-kafka.version>
+		<spring-cloud-stream.version>1.1.1.RELEASE</spring-cloud-stream.version>
 	</properties>
 	<modules>
 		<module>spring-cloud-stream-binder-kafka</module>
 		<module>spring-cloud-starter-stream-kafka</module>
-		<module>spring-cloud-stream-binder-kafka-test-support</module>
 		<module>spring-cloud-stream-binder-kafka-docs</module>
+		<module>spring-cloud-stream-binder-kafka-0.10-test</module>
 	</modules>
+
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-codec</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.apache.kafka</groupId>
+				<artifactId>kafka_2.11</artifactId>
+				<version>${kafka.version}</version>
+				<exclusions>
+					<exclusion>
+						<groupId>org.slf4j</groupId>
+						<artifactId>slf4j-log4j12</artifactId>
+					</exclusion>
+					<exclusion>
+						<groupId>log4j</groupId>
+						<artifactId>log4j</artifactId>
+					</exclusion>
+				</exclusions>
+			</dependency>
+			<dependency>
+				<groupId>org.apache.kafka</groupId>
+				<artifactId>kafka-clients</artifactId>
+				<version>${kafka.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.kafka</groupId>
+				<artifactId>spring-kafka</artifactId>
+				<version>${spring-kafka.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.integration</groupId>
+				<artifactId>spring-integration-kafka</artifactId>
+				<version>${spring-integration-kafka.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-binder-test</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+				<scope>test</scope>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.kafka</groupId>
+				<artifactId>spring-kafka-test</artifactId>
+				<scope>test</scope>
+				<version>${spring-kafka.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>org.apache.kafka</groupId>
+				<artifactId>kafka_2.11</artifactId>
+				<classifier>test</classifier>
+				<version>${kafka.version}</version>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+	
 	<build>
 		<pluginManagement>
 			<plugins>
-				<plugin>
-					<groupId>org.apache.maven.plugins</groupId>
-					<artifactId>maven-surefire-plugin</artifactId>
-					<configuration>
-						<redirectTestOutputToFile>true</redirectTestOutputToFile>
-					</configuration>
-				</plugin>
 				<plugin>
 					<groupId>org.apache.maven.plugins</groupId>
 					<artifactId>maven-antrun-plugin</artifactId>
@@ -45,7 +103,7 @@
 						<dependency>
 							<groupId>com.puppycrawl.tools</groupId>
 							<artifactId>checkstyle</artifactId>
-							<version>6.17</version>
+							<version>7.1</version>
 						</dependency>
 					</dependencies>
 				</plugin>
@@ -56,9 +114,46 @@
 						<quiet>true</quiet>
 					</configuration>
 				</plugin>
+				<plugin>
+					<groupId>org.apache.maven.plugins</groupId>
+					<artifactId>maven-surefire-plugin</artifactId>
+					<configuration>
+						<redirectTestOutputToFile>true</redirectTestOutputToFile>
+					</configuration>
+				</plugin>
 			</plugins>
 		</pluginManagement>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-checkstyle-plugin</artifactId>
+				<dependencies>
+					<dependency>
+						<groupId>org.springframework.cloud</groupId>
+						<artifactId>spring-cloud-build-tools</artifactId>
+						<version>1.2.0.RELEASE</version>
+					</dependency>
+				</dependencies>
+				<executions>
+					<execution>
+						<id>checkstyle-validation</id>
+						<phase>validate</phase>
+						<configuration>
+							<configLocation>checkstyle.xml</configLocation>
+							<encoding>UTF-8</encoding>
+							<consoleOutput>true</consoleOutput>
+							<failsOnError>true</failsOnError>
+							<includeTestSourceDirectory>true</includeTestSourceDirectory>
+						</configuration>
+						<goals>
+							<goal>check</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
 	</build>
+
 	<profiles>
 		<profile>
 			<id>spring</id>
--- a/spring-cloud-starter-stream-kafka/pom.xml
+++ b/spring-cloud-starter-stream-kafka/pom.xml
@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.1.0.M1</version>
+		<version>1.1.1.RELEASE</version>
 	</parent>
 	<artifactId>spring-cloud-starter-stream-kafka</artifactId>
 	<description>Spring Cloud Starter Stream Kafka</description>
@@ -20,7 +20,7 @@
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
-			<version>1.1.0.M1</version>
+			<version>1.1.1.RELEASE</version>
 		</dependency>
 	</dependencies>
 </project>
--- a/spring-cloud-stream-binder-kafka-0.10-test/pom.xml
+++ b/spring-cloud-stream-binder-kafka-0.10-test/pom.xml
@@ -0,0 +1,104 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
+		<version>1.1.1.RELEASE</version>
+	</parent>
+	<artifactId>spring-cloud-stream-binder-kafka-0.10-test</artifactId>
+	<description>Spring Cloud Stream Kafka Binder 0.10 Tests</description>
+	<url>http://projects.spring.io/spring-cloud</url>
+	<organization>
+		<name>Pivotal Software, Inc.</name>
+		<url>http://www.spring.io</url>
+	</organization>
+	<properties>
+		<main.basedir>${basedir}/../..</main.basedir>
+		<!--
+		   Override Kafka dependencies to Kafka 0.10 and supporting Spring Kafka and
+		   Spring Integration Kafka versions
+		-->
+		<kafka.version>0.10.0.0</kafka.version>
+		<spring-kafka.version>1.1.1.RELEASE</spring-kafka.version>
+		<spring-integration-kafka.version>2.1.0.RELEASE</spring-integration-kafka.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.11</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-log4j12</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-clients</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-schema</artifactId>
+			<version>${spring-cloud-stream.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.confluent</groupId>
+			<artifactId>kafka-avro-serializer</artifactId>
+			<version>3.0.1</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.confluent</groupId>
+			<artifactId>kafka-schema-registry</artifactId>
+			<version>3.0.1</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<repositories>
+		<repository>
+			<id>confluent</id>
+			<url>http://packages.confluent.io/maven/</url>
+		</repository>
+	</repositories>
+
+
+</project>
--- a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java
+++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java
@@ -0,0 +1,245 @@
+/*
+ * Copyright 2014-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
+import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
+import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+import org.I0Itec.zkclient.ZkClient;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.eclipse.jetty.server.Server;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.Spy;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.core.BrokerAddress;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.retry.RetryOperations;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Integration tests for the {@link KafkaMessageChannelBinder}.
+ *
+ * @author Eric Bottard
+ * @author Marius Bogoevici
+ * @author Mark Fisher
+ * @author Ilayaperumal Gopinathan
+ */
+public class Kafka10BinderTests extends KafkaBinderTests {
+
+	private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10);
+
+	private Kafka10TestBinder binder;
+
+	private Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
+
+	@Override
+	protected void binderBindUnbindLatency() throws InterruptedException {
+		Thread.sleep(500);
+	}
+
+	@Override
+	protected Kafka10TestBinder getBinder() {
+		if (binder == null) {
+			KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
+			binder = new Kafka10TestBinder(binderConfiguration);
+		}
+		return binder;
+	}
+
+	protected KafkaBinderConfigurationProperties createConfigurationProperties() {
+		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
+		BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
+		List<String> bAddresses = new ArrayList<>();
+		for (BrokerAddress bAddress : brokerAddresses) {
+			bAddresses.add(bAddress.toString());
+		}
+		String[] foo = new String[bAddresses.size()];
+		binderConfiguration.setBrokers(bAddresses.toArray(foo));
+		binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString());
+		return binderConfiguration;
+	}
+
+	@Override
+	protected int partitionSize(String topic) {
+		return consumerFactory().createConsumer().partitionsFor(topic).size();
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected void setMetadataRetryOperations(Binder binder, RetryOperations retryOperations) {
+		((Kafka10TestBinder) binder).getBinder().setMetadataRetryOperations(retryOperations);
+	}
+
+	@Override
+	protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
+		final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(),
+				kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+
+		return new ZkUtils(zkClient, null, false);
+	}
+
+	@Override
+	protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) {
+		adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties());
+	}
+
+	@Override
+	protected int invokePartitionSize(String topic, ZkUtils zkUtils) {
+		return adminUtilsOperation.partitionSize(topic, zkUtils);
+	}
+
+	@Override
+	public String getKafkaOffsetHeaderKey() {
+		return KafkaHeaders.OFFSET;
+	}
+
+	@Override
+	protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
+		return new Kafka10TestBinder(kafkaBinderConfigurationProperties);
+	}
+
+	@Before
+	public void init() {
+		String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER");
+		if (multiplier != null) {
+			timeoutMultiplier = Double.parseDouble(multiplier);
+		}
+	}
+
+	@Override
+	protected boolean usesExplicitRouting() {
+		return false;
+	}
+
+	@Override
+	protected String getClassUnderTestName() {
+		return CLASS_UNDER_TEST_NAME;
+	}
+
+	@Override
+	public Spy spyOn(final String name) {
+		throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests");
+	}
+
+
+	private ConsumerFactory<byte[], byte[]> consumerFactory() {
+		Map<String, Object> props = new HashMap<>();
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, "TEST-CONSUMER-GROUP");
+		Deserializer<byte[]> valueDecoder = new ByteArrayDeserializer();
+		Deserializer<byte[]> keyDecoder = new ByteArrayDeserializer();
+
+		return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder);
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testCustomAvroSerialization() throws Exception {
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+		final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
+				configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+		final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+		Map<String, Object> schemaRegistryProps = new HashMap<>();
+		schemaRegistryProps.put("kafkastore.connection.url", configurationProperties.getZkConnectionString());
+		schemaRegistryProps.put("listeners", "http://0.0.0.0:8082");
+		schemaRegistryProps.put("port", "8082");
+		schemaRegistryProps.put("kafkastore.topic", "_schemas");
+		SchemaRegistryConfig config = new SchemaRegistryConfig(schemaRegistryProps);
+		SchemaRegistryRestApplication app = new SchemaRegistryRestApplication(config);
+		Server server = app.createServer();
+		server.start();
+		long endTime = System.currentTimeMillis() + 5000;
+		while(true) {
+			if (server.isRunning()) {
+				break;
+			}
+			else if (System.currentTimeMillis() > endTime) {
+				fail("Kafka Schema Registry Server failed to start");
+			}
+		}
+		User1 firstOutboundFoo = new User1();
+		String userName1 = "foo-name" + UUID.randomUUID().toString();
+		String favColor1 = "foo-color" + UUID.randomUUID().toString();
+		firstOutboundFoo.setName(userName1);
+		firstOutboundFoo.setFavoriteColor(favColor1);
+		Message<?> message = MessageBuilder.withPayload(firstOutboundFoo).build();
+		SubscribableChannel moduleOutputChannel = new DirectChannel();
+		String testTopicName = "existing" + System.currentTimeMillis();
+		invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties());
+		configurationProperties.setAutoAddPartitions(true);
+		Binder binder = getBinder(configurationProperties);
+		QueueChannel moduleInputChannel = new QueueChannel();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.getExtension().getConfiguration().put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
+		producerProperties.getExtension().getConfiguration().put("schema.registry.url", "http://localhost:8082");
+		producerProperties.setUseNativeEncoding(true);
+		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+		consumerProperties.getExtension().getConfiguration().put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
+		consumerProperties.getExtension().getConfiguration().put("schema.registry.url", "http://localhost:8082");
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = receive(moduleInputChannel);
+		assertThat(inbound).isNotNull();
+		assertTrue(message.getPayload() instanceof User1);
+		User1 receivedUser = (User1) message.getPayload();
+		assertThat(receivedUser.getName()).isEqualTo(userName1);
+		assertThat(receivedUser.getFavoriteColor()).isEqualTo(favColor1);
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+}
--- a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java
+++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2015-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.kafka.support.LoggingProducerListener;
+import org.springframework.kafka.support.ProducerListener;
+
+/**
+ * Test support class for {@link KafkaMessageChannelBinder}.
+ * @author Eric Bottard
+ * @author Marius Bogoevici
+ * @author David Turanski
+ * @author Gary Russell
+ * @author Soby Chacko
+ */
+public class Kafka10TestBinder extends AbstractKafkaTestBinder {
+
+	public Kafka10TestBinder(KafkaBinderConfigurationProperties binderConfiguration) {
+		try {
+			KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration);
+			binder.setCodec(getCodec());
+			ProducerListener producerListener = new LoggingProducerListener();
+			binder.setProducerListener(producerListener);
+			GenericApplicationContext context = new GenericApplicationContext();
+			context.refresh();
+			binder.setApplicationContext(context);
+			binder.setAdminUtilsOperation(new Kafka10AdminUtilsOperation());
+			binder.afterPropertiesSet();
+			this.setBinder(binder);
+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+}
--- a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java
+++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.io.IOException;
+
+import org.apache.avro.Schema;
+import org.apache.avro.reflect.Nullable;
+import org.apache.avro.specific.SpecificRecordBase;
+
+import org.springframework.core.io.ClassPathResource;
+
+/**
+ * @author Marius Bogoevici
+ * @author Ilayaperumal Gopinathan
+ */
+public class User1 extends SpecificRecordBase {
+
+	@Nullable
+	private String name;
+
+	@Nullable
+	private String favoriteColor;
+
+	public String getName() {
+		return this.name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getFavoriteColor() {
+		return this.favoriteColor;
+	}
+
+	public void setFavoriteColor(String favoriteColor) {
+		this.favoriteColor = favoriteColor;
+	}
+
+	@Override
+	public Schema getSchema() {
+		try {
+			return new Schema.Parser().parse(new ClassPathResource("schemas/users_v1.schema").getInputStream());
+		}
+		catch (IOException e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
+	@Override
+	public Object get(int i) {
+		if (i == 0) {
+			return getName().toString();
+		}
+		if (i == 1) {
+			return getFavoriteColor().toString();
+		}
+		return null;
+	}
+
+	@Override
+	public void put(int i, Object o) {
+		if (i == 0) {
+			setName((String) o);
+		}
+		if (i == 1) {
+			setFavoriteColor((String) o);
+		}
+	}
+}
--- a/spring-cloud-stream-binder-kafka-0.10-test/src/test/resources/schemas/users_v1.schema
+++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/resources/schemas/users_v1.schema
@@ -0,0 +1,8 @@
+{"namespace": "org.springframework.cloud.stream.binder.kafka",
+  "type": "record",
+  "name": "User1",
+  "fields": [
+    {"name": "name", "type": "string"},
+    {"name": "favoriteColor", "type": "string"}
+  ]
+}
--- a/spring-cloud-stream-binder-kafka-docs/pom.xml
+++ b/spring-cloud-stream-binder-kafka-docs/pom.xml
@@ -5,7 +5,7 @@
    <parent>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.1.0.M1</version>
+        <version>1.1.1.RELEASE</version>
    </parent>

    <artifactId>spring-cloud-stream-binder-kafka-docs</artifactId>
@@ -18,7 +18,7 @@
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka</artifactId>
-            <version>1.1.0.M1</version>
+            <version>${project.version}</version>
        </dependency>
    </dependencies>
    <profiles>
--- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc
+++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc
@@ -120,9 +120,19 @@ Default: `2097152`.
 The following properties are available for Kafka consumers only and
 must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.consumer.`.

+autoRebalanceEnabled::
+When `true`, topic partitions will be automatically rebalanced between the members of a consumer group.
+When `false`, each consumer will be assigned a fixed set of partitions based on `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex`.
+This requires both `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex` properties to be set appropriately on each launched instance.
+The property `spring.cloud.stream.instanceCount` must typically be greater than 1 in this case.
+
+Default: `true`.
 autoCommitOffset::
  Whether to autocommit offsets when a message has been processed.
-If set to `false`, an `Acknowledgment` header will be available in the message headers for late acknowledgment.
+If set to `false`, a header with the key `kafka_acknowledgment` of the type `org.springframework.kafka.support.Acknowledgment` header will be present in the inbound message.
+Applications may use this header for acknowledging messages.
+See the examples section for details.
+When this property is set to `false`, Kafka binder will set the ack mode to `org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL`.
 +
 Default: `true`.
 autoCommitOnError::
@@ -192,6 +202,34 @@ If a topic already exists with a larger number of partitions than the maximum of

 In this section, we illustrate the use of the above properties for specific scenarios.

+==== Example: Setting `autoCommitOffset` false and relying on manual acking.
+
+This example illustrates how one may manually acknowledge offsets in a consumer application.
+
+This example requires that `spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset` is set to false.
+Use the corresponding input channel name for your example.
+
+[source]
+----
+@SpringBootApplication
+@EnableBinding(Sink.class)
+public class ManuallyAcknowdledgingConsumer {
+
+ public static void main(String[] args) {
+     SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args);
+ }
+
+ @StreamListener(Sink.INPUT)
+ public void process(Message<?> message) {
+     Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
+     if (acknowledgment != null) {
+         System.out.println("Acknowledgment provided");
+         acknowledgment.acknowledge();
+     }
+ }
+}
+----
+
 ==== Example: security configuration

 Apache Kafka 0.9 supports secure connections between client and brokers.
@@ -208,22 +246,150 @@ 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.
-At the time of this release, the JAAS, and (optionally) krb5 file locations must be set for Spring Cloud Stream applications by using system properties.
-Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos.
+
+Spring Cloud Stream supports passing JAAS configuration information to the application using a JAAS configuration file and using Spring Boot properties.
+
+===== Using JAAS configuration files
+
+The JAAS, and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties.
+Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos using a JAAS configuration file:

 [source]
 ----
- java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \\
-   --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \\
-   --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \\
-   --spring.cloud.stream.bindings.input.destination=stream.ticktock \\
-   --spring.cloud.stream.kafka.binder.clientConfiguration.security.protocol=SASL_PLAINTEXT
+ java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \
+   --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
+   --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \
+   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
+   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT
 ----

+===== Using Spring Boot properties
+
+As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications using Spring Boot properties.
+
+The following properties can be used for configuring the login context of the Kafka client.
+
+spring.cloud.stream.kafka.binder.jaas.loginModule::
+  The login module name. Not necessary to be set in normal cases.
+
+Default: `com.sun.security.auth.module.Krb5LoginModule`.
+spring.cloud.stream.kafka.binder.jaas.controlFlag::
+  The control flag of the login module.
+
+Default: `required`.
+spring.cloud.stream.kafka.binder.jaas.options::
+  Map with a key/value pair containing the login module options.
+
+Default: Empty map.
+
+Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos using Spring Boot configuration properties:
+
+[source]
+----
+ java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
+   --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \
+   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
+   --spring.cloud.stream.kafka.binder.autoCreateTopics=false \
+   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \
+   --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \
+   --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \
+   --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \
+   --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM
+----
+
+This represents the equivalent of the following JAAS file:
+
+[source]
+----
+KafkaClient {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/security/keytabs/kafka_client.keytab"
+    principal="kafka-client-1@EXAMPLE.COM";
+};
+----
+
+If the topics required already exist on the broker, or will be created by an administrator, autocreation can be turned off and only client JAAS properties need to be sent. As an alternative to setting `spring.cloud.stream.kafka.binder.autoCreateTopics` you can simply remove the broker dependency from the application. See <<exclude-admin-utils>> for details.
+
+[NOTE]
+====
+Do not mix JAAS configuration files and Spring Boot properties in the same application.
+If the `-Djava.security.auth.login.config` system property is already present, Spring Cloud Stream will ignore the Spring Boot properties.
+
+====

 [NOTE]
 ====
 Exercise caution when using the `autoCreateTopics` and `autoAddPartitions` if using Kerberos.
 Usually applications may use principals that do not have administrative rights in Kafka and Zookeeper, and relying on Spring Cloud Stream to create/modify topics may fail.
 In secure environments, we strongly recommend creating topics and managing ACLs administratively using Kafka tooling.
-====
+====
+
+==== Using the binder with Apache Kafka 0.10
+
+The binder also supports connecting to Kafka 0.10 brokers.
+In order to support this, when you create the project that contains your application, include `spring-cloud-starter-stream-kafka` as you normally would do for 0.9 based applications.
+Then add these dependencies at the top of the `<dependencies>` section in the pom.xml file to override the Apache Kafka, Spring Kafka, and Spring Integration Kafka with 0.10-compatible versions as in the following example:
+
+[source,xml]
+----
+<dependency>
+  <groupId>org.springframework.kafka</groupId>
+  <artifactId>spring-kafka</artifactId>
+  <version>1.1.1.RELEASE</version>
+</dependency>
+<dependency>
+  <groupId>org.springframework.integration</groupId>
+  <artifactId>spring-integration-kafka</artifactId>
+  <version>2.1.0.RELEASE</version>
+</dependency>
+<dependency>
+  <groupId>org.apache.kafka</groupId>
+  <artifactId>kafka_2.11</artifactId>
+  <version>0.10.0.0</version>
+  <exclusions>
+    <exclusion>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+    </exclusion>
+  </exclusions>
+</dependency>
+----
+
+[NOTE]
+====
+The versions above are provided only for the sake of the example.
+For best results, we recommend using the most recent 0.10-compatible versions of the projects.
+====
+
+[[exclude-admin-utils]]
+==== Excluding Kafka broker jar from the classpath of the binder based application
+
+The Apache Kafka Binder uses the administrative utilities which are part of the Apache Kafka server library to create and reconfigure topics.
+If the inclusion of the Apache Kafka server library and its dependencies is not necessary at runtime because the application will rely on the topics being configured administratively, the Kafka binder allows for Apache Kafka server dependency to be excluded from the application.
+
+If you use Kafka 10 dependencies as advised above, all you have to do is not to include the kafka broker dependency.
+If you use Kafka 0.9, then ensure that you exclude the kafka broker jar from the `spring-cloud-starter-stream-kafka` dependency as following.
+
+[source,xml]
+----
+<dependency>
+  <groupId>org.springframework.cloud</groupId>
+  <artifactId>spring-cloud-starter-stream-kafka</artifactId>
+  <exclusions>
+    <exclusion>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka_2.11</artifactId>
+    </exclusion>
+  </exclusions>
+</dependency>
+----
+
+If you exclude the Apache Kafka server dependency and the topic is not present on the server, then the Apache Kafka broker will create the topic if auto topic creation is enabled on the server.
+Please keep in mind that if you are relying on this, then the Kafka server will use the default number of partitions and replication factors.
+On the other hand, if auto topic creation is disabled on the server, then care must be taken before running the application to create the topic with the desired number of partitions.
+
+If you want to have full control over how partitions are allocated, then leave the default settings as they are, i.e. do not exclude the kafka broker jar and ensure that `spring.cloud.stream.kafka.binder.autoCreateTopics` is set to `true`, which is the default.
+
+
--- a/spring-cloud-stream-binder-kafka-test-support/pom.xml
+++ b/spring-cloud-stream-binder-kafka-test-support/pom.xml
@@ -1,24 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <groupId>org.springframework.cloud</groupId>
-        <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-        <version>1.1.0.M1</version>
-    </parent>
-    <artifactId>spring-cloud-stream-binder-kafka-test-support</artifactId>
-    <description>Kafka related test classes</description>
-
-    <dependencies>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.kafka</groupId>
-            <artifactId>spring-kafka-test</artifactId>
-            <version>${spring-kafka.version}</version>
-        </dependency>
-    </dependencies>
-</project>
--- a/spring-cloud-stream-binder-kafka/pom.xml
+++ b/spring-cloud-stream-binder-kafka/pom.xml
@@ -10,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>1.1.0.M1</version>
+		<version>1.1.1.RELEASE</version>
 	</parent>

 	<dependencies>
@@ -37,12 +37,6 @@
 			<artifactId>spring-cloud-stream-binder-test</artifactId>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-kafka-test-support</artifactId>
-			<scope>test</scope>
-			<version>1.1.0.M1</version>
-		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-kafka</artifactId>
@@ -54,17 +48,6 @@
 				</exclusion>
 			</exclusions>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.kafka</groupId>
-			<artifactId>spring-kafka</artifactId>
-			<version>${spring-kafka.version}</version>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework.kafka</groupId>
-			<artifactId>spring-kafka-test</artifactId>
-			<scope>test</scope>
-			<version>${spring-kafka.version}</version>
-		</dependency>
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka_2.11</artifactId>
@@ -73,6 +56,26 @@
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka-clients</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>${spring-kafka.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-kafka</artifactId>
+			<version>${spring-integration-kafka.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka_2.11</artifactId>
@@ -81,34 +84,20 @@
 		</dependency>
 	</dependencies>

-	<dependencyManagement>
-		<dependencies>
-			<dependency>
-				<groupId>org.apache.kafka</groupId>
-				<artifactId>kafka_2.11</artifactId>
-				<version>${kafka.version}</version>
-				<exclusions>
-					<exclusion>
-						<groupId>org.slf4j</groupId>
-						<artifactId>slf4j-log4j12</artifactId>
-					</exclusion>
-					<exclusion>
-						<groupId>log4j</groupId>
-						<artifactId>log4j</artifactId>
-					</exclusion>
-				</exclusions>
-			</dependency>
-			<dependency>
-				<groupId>org.apache.kafka</groupId>
-				<artifactId>kafka_2.11</artifactId>
-				<classifier>test</classifier>
-				<version>${kafka.version}</version>
-			</dependency>
-			<dependency>
-				<groupId>org.apache.kafka</groupId>
-				<artifactId>kafka-clients</artifactId>
-				<version>${kafka.version}</version>
-			</dependency>
-		</dependencies>
-	</dependencyManagement>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<version>3.0.2</version>
+				<executions>
+					<execution>
+						<goals>
+							<goal>test-jar</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
 </project>
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
@@ -33,6 +33,7 @@ import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigura

 /**
 * Health indicator for Kafka.
+ *
 * @author Ilayaperumal Gopinathan
 * @author Marius Bogoevici
 */
@@ -43,7 +44,7 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
 	private final KafkaBinderConfigurationProperties configurationProperties;

 	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
-			KafkaBinderConfigurationProperties configurationProperties) {
+										KafkaBinderConfigurationProperties configurationProperties) {
 		this.binder = binder;
 		this.configurationProperties = configurationProperties;

--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java
@@ -0,0 +1,118 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+
+import org.apache.kafka.common.security.JaasUtils;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.util.Assert;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KafkaBinderJaasInitializerListener implements ApplicationListener<ContextRefreshedEvent>,
+		ApplicationContextAware, DisposableBean {
+
+	public static final String DEFAULT_ZK_LOGIN_CONTEXT_NAME = "Client";
+
+	private ApplicationContext applicationContext;
+
+	private final boolean ignoreJavaLoginConfigParamSystemProperty;
+
+	private final File placeholderJaasConfiguration;
+
+	public KafkaBinderJaasInitializerListener() throws IOException {
+		// we ignore the system property if it wasn't originally set at launch
+		this.ignoreJavaLoginConfigParamSystemProperty =
+				(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) == null);
+		this.placeholderJaasConfiguration = File.createTempFile("kafka-client-jaas-config-placeholder", "conf");
+		this.placeholderJaasConfiguration.deleteOnExit();
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.applicationContext = applicationContext;
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		if (this.ignoreJavaLoginConfigParamSystemProperty) {
+			System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
+		}
+	}
+
+	@Override
+	public void onApplicationEvent(ContextRefreshedEvent event) {
+		if (event.getSource() == this.applicationContext) {
+			KafkaBinderConfigurationProperties binderConfigurationProperties =
+					applicationContext.getBean(KafkaBinderConfigurationProperties.class);
+			// only use programmatic support if a file is not set via system property
+			if (ignoreJavaLoginConfigParamSystemProperty
+					&& binderConfigurationProperties.getJaas() != null) {
+				Map<String, AppConfigurationEntry[]> configurationEntries = new HashMap<>();
+				AppConfigurationEntry kafkaClientConfigurationEntry = new AppConfigurationEntry
+						(binderConfigurationProperties.getJaas().getLoginModule(),
+								binderConfigurationProperties.getJaas().getControlFlagValue(),
+								binderConfigurationProperties.getJaas().getOptions() != null ?
+										binderConfigurationProperties.getJaas().getOptions() :
+										Collections.<String, Object>emptyMap());
+				configurationEntries.put(JaasUtils.LOGIN_CONTEXT_CLIENT,
+						new AppConfigurationEntry[]{ kafkaClientConfigurationEntry });
+				Configuration.setConfiguration(new InternalConfiguration(configurationEntries));
+				// Workaround for a 0.9 client issue where even if the Configuration is set
+				// a system property check is performed.
+				// Since the Configuration already exists, this will be ignored.
+				if (this.placeholderJaasConfiguration != null) {
+					System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, this.placeholderJaasConfiguration.getAbsolutePath());
+				}
+			}
+		}
+	}
+
+	/**
+	 * A {@link Configuration} set up programmatically by the Kafka binder
+	 */
+	public static class InternalConfiguration extends Configuration {
+
+		private final Map<String, AppConfigurationEntry[]> configurationEntries;
+
+		public InternalConfiguration(Map<String, AppConfigurationEntry[]> configurationEntries) {
+			Assert.notNull(configurationEntries, " cannot be null");
+			Assert.notEmpty(configurationEntries, " cannot be empty");
+			this.configurationEntries = configurationEntries;
+		}
+
+		@Override
+		public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+			return configurationEntries.get(name);
+		}
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -25,8 +25,6 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;

-import kafka.admin.AdminUtils;
-import kafka.api.TopicMetadata;
 import kafka.common.ErrorMapping;
 import kafka.utils.ZkUtils;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -40,9 +38,7 @@ import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.security.JaasUtils;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.Utils;
-import scala.collection.Seq;

 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -52,6 +48,7 @@ import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation;
 import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.common.LiteralExpression;
@@ -63,6 +60,7 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ErrorHandler;
 import org.springframework.kafka.listener.config.ContainerProperties;
@@ -83,6 +81,7 @@ import org.springframework.util.StringUtils;

 /**
 * A {@link Binder} that uses Kafka as the underlying middleware.
+ *
 * @author Eric Bottard
 * @author Marius Bogoevici
 * @author Ilayaperumal Gopinathan
@@ -109,6 +108,8 @@ public class KafkaMessageChannelBinder extends

 	private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();

+	private AdminUtilsOperation adminUtilsOperation;
+
 	public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties) {
 		super(false, headersToMap(configurationProperties));
 		this.configurationProperties = configurationProperties;
@@ -130,8 +131,13 @@ public class KafkaMessageChannelBinder extends
 		return headersToMap;
 	}

+	public void setAdminUtilsOperation(AdminUtilsOperation adminUtilsOperation) {
+		this.adminUtilsOperation = adminUtilsOperation;
+	}
+
 	/**
 	 * Retry configuration for operations such as validating topic creation
+	 *
 	 * @param metadataRetryOperations the retry configuration
 	 */
 	public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) {
@@ -189,11 +195,11 @@ public class KafkaMessageChannelBinder extends

 	@Override
 	protected MessageHandler createProducerMessageHandler(final String destination,
-			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
+															ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {

 		KafkaTopicUtils.validateTopicName(destination);
-
-		Collection<PartitionInfo> partitions = ensureTopicCreated(destination, producerProperties.getPartitionCount());
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(destination, producerProperties.getPartitionCount());
+		Collection<PartitionInfo> partitions = getPartitionsForTopic(destination, producerProperties.getPartitionCount());

 		if (producerProperties.getPartitionCount() < partitions.size()) {
 			if (this.logger.isInfoEnabled()) {
@@ -215,12 +221,13 @@ public class KafkaMessageChannelBinder extends

 	@Override
 	protected String createProducerDestinationIfNecessary(String name,
-			ExtendedProducerProperties<KafkaProducerProperties> properties) {
+														ExtendedProducerProperties<KafkaProducerProperties> properties) {
 		if (this.logger.isInfoEnabled()) {
 			this.logger.info("Using kafka topic for outbound: " + name);
 		}
 		KafkaTopicUtils.validateTopicName(name);
-		Collection<PartitionInfo> partitions = ensureTopicCreated(name, properties.getPartitionCount());
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount());
+		Collection<PartitionInfo> partitions = getPartitionsForTopic(name, properties.getPartitionCount());
 		if (properties.getPartitionCount() < partitions.size()) {
 			if (this.logger.isInfoEnabled()) {
 				this.logger.info("The `partitionCount` of the producer for topic " + name + " is "
@@ -240,8 +247,7 @@ public class KafkaMessageChannelBinder extends
 		}
 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
 		props.put(ProducerConfig.RETRIES_CONFIG, 0);
-		props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
-		props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
+		props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize()));
 		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
@@ -250,7 +256,6 @@ public class KafkaMessageChannelBinder extends
 				String.valueOf(producerProperties.getExtension().getBatchTimeout()));
 		props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
 				producerProperties.getExtension().getCompressionType().toString());
-
 		if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) {
 			props.putAll(producerProperties.getExtension().getConfiguration());
 		}
@@ -259,17 +264,19 @@ public class KafkaMessageChannelBinder extends

 	@Override
 	protected Collection<PartitionInfo> createConsumerDestinationIfNecessary(String name, String group,
-			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+																			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 		KafkaTopicUtils.validateTopicName(name);
 		if (properties.getInstanceCount() == 0) {
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
-		Collection<PartitionInfo> allPartitions = ensureTopicCreated(name,
-				properties.getInstanceCount() * properties.getConcurrency());
+		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount);
+		Collection<PartitionInfo> allPartitions = getPartitionsForTopic(name, partitionCount);

 		Collection<PartitionInfo> listenedPartitions;

-		if (properties.getInstanceCount() == 1) {
+		if (properties.getExtension().isAutoRebalanceEnabled() ||
+				properties.getInstanceCount() == 1) {
 			listenedPartitions = allPartitions;
 		}
 		else {
@@ -288,34 +295,25 @@ public class KafkaMessageChannelBinder extends
 	@Override
 	@SuppressWarnings("unchecked")
 	protected MessageProducer createConsumerEndpoint(String name, String group, Collection<PartitionInfo> destination,
-			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+													ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(),
 				"DLQ support is not available for anonymous subscriptions");
 		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
-
 		Map<String, Object> props = getConsumerConfig(anonymous, consumerGroup);
-		Deserializer<byte[]> valueDecoder = new ByteArrayDeserializer();
-		Deserializer<byte[]> keyDecoder = new ByteArrayDeserializer();
-
 		if (!ObjectUtils.isEmpty(properties.getExtension().getConfiguration())) {
 			props.putAll(properties.getExtension().getConfiguration());
 		}
-
-		ConsumerFactory<byte[], byte[]> consumerFactory = new DefaultKafkaConsumerFactory<>(props, keyDecoder,
-				valueDecoder);
-
+		ConsumerFactory<?, ?> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
 		Collection<PartitionInfo> listenedPartitions = destination;
 		Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
 		final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
 				listenedPartitions);
-
 		final ContainerProperties containerProperties =
 				anonymous || properties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(name)
 						: new ContainerProperties(topicPartitionInitialOffsets);
-
 		int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size());
-		final ConcurrentMessageListenerContainer<byte[], byte[]> messageListenerContainer =
+		final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
 				new ConcurrentMessageListenerContainer(
 						consumerFactory, containerProperties) {

@@ -326,25 +324,23 @@ public class KafkaMessageChannelBinder extends
 				};
 		messageListenerContainer.setConcurrency(concurrency);
 		messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(properties));
-
+		if (!properties.getExtension().isAutoCommitOffset()) {
+			messageListenerContainer.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
+		}
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug(
 					"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
 		}
-
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug(
 					"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
 		}
-
-		final KafkaMessageDrivenChannelAdapter<byte[], byte[]> kafkaMessageDrivenChannelAdapter =
+		final KafkaMessageDrivenChannelAdapter<?, ?> kafkaMessageDrivenChannelAdapter =
 				new KafkaMessageDrivenChannelAdapter<>(
 						messageListenerContainer);
-
 		kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
 		final RetryTemplate retryTemplate = buildRetryTemplate(properties);
 		kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate);
-
 		if (properties.getExtension().isEnableDlq()) {
 			final String dlqTopic = "error." + name + "." + group;
 			initDlqProducer();
@@ -387,6 +383,8 @@ public class KafkaMessageChannelBinder extends

 	private Map<String, Object> getConsumerConfig(boolean anonymous, String consumerGroup) {
 		Map<String, Object> props = new HashMap<>();
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
 		if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
 			props.putAll(configurationProperties.getConfiguration());
 		}
@@ -418,94 +416,104 @@ public class KafkaMessageChannelBinder extends
 		return topicPartitionInitialOffsets;
 	}

+	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount) {
+		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
+			createTopicAndPartitions(topicName, partitionCount);
+		}
+		else if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation == null) {
+			this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " +
+					"No topic will be created by the binder");
+		}
+		else if (!this.configurationProperties.isAutoCreateTopics()) {
+			this.logger.info("Auto creation of topics is disabled.");
+		}
+	}
+
 	/**
 	 * Creates a Kafka topic if needed, or try to increase its partition count to the
 	 * desired number.
 	 */
-	private Collection<PartitionInfo> ensureTopicCreated(final String topicName, final int partitionCount) {
+	private void createTopicAndPartitions(final String topicName, final int partitionCount) {

 		final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 				this.configurationProperties.getZkSessionTimeout(),
 				this.configurationProperties.getZkConnectionTimeout(),
 				JaasUtils.isZkSecurityEnabled());
 		try {
-			final Properties topicConfig = new Properties();
-			TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkUtils);
-			if (topicMetadata.errorCode() == ErrorMapping.NoError()) {
-				// only consider minPartitionCount for resizing if autoAddPartitions is
-				// true
+			short errorCode = adminUtilsOperation.errorCodeFromTopicMetadata(topicName, zkUtils);
+			if (errorCode == ErrorMapping.NoError()) {
+				// only consider minPartitionCount for resizing if autoAddPartitions is true
 				int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions()
 						? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount)
 						: partitionCount;
-				if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) {
+				int partitionSize = adminUtilsOperation.partitionSize(topicName, zkUtils);
+
+				if (partitionSize < effectivePartitionCount) {
 					if (this.configurationProperties.isAutoAddPartitions()) {
-						AdminUtils.addPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
+						adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
 					}
 					else {
-						int topicSize = topicMetadata.partitionsMetadata().size();
 						throw new BinderException("The number of expected partitions was: " + partitionCount + ", but "
-								+ topicSize + (topicSize > 1 ? " have " : " has ") + "been found instead."
+								+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
 								+ "Consider either increasing the partition count of the topic or enabling " +
 								"`autoAddPartitions`");
 					}
 				}
 			}
-			else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) {
-				if (this.configurationProperties.isAutoCreateTopics()) {
-					Seq<Object> brokerList = zkUtils.getSortedBrokerList();
-					// always consider minPartitionCount for topic creation
-					final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(),
-							partitionCount);
-					this.metadataRetryOperations.execute(new RetryCallback<Object, RuntimeException>() {
+			else if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode()) {
+				// always consider minPartitionCount for topic creation
+				final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(),
+						partitionCount);

-						@Override
-						public Object doWithRetry(RetryContext context) throws RuntimeException {
-							AdminUtils.createTopic(zkUtils, topicName, effectivePartitionCount,
-									configurationProperties.getReplicationFactor(), new Properties());
-							return null;
-						}
-					});
-				}
-				else {
-					throw new BinderException("Topic " + topicName + " does not exist");
-				}
+				this.metadataRetryOperations.execute(new RetryCallback<Object, RuntimeException>() {
+
+					@Override
+					public Object doWithRetry(RetryContext context) throws RuntimeException {
+
+						adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount,
+								configurationProperties.getReplicationFactor(), new Properties());
+						return null;
+					}
+				});
 			}
 			else {
 				throw new BinderException("Error fetching Kafka topic metadata: ",
-						ErrorMapping.exceptionFor(topicMetadata.errorCode()));
+						ErrorMapping.exceptionFor(errorCode));
 			}
-			try {
-				return this.metadataRetryOperations
-						.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
-
-							@Override
-							public Collection<PartitionInfo> doWithRetry(RetryContext context) throws Exception {
-								Collection<PartitionInfo> partitions =
-										getProducerFactory(
-												new ExtendedProducerProperties<>(new KafkaProducerProperties()))
-												.createProducer().partitionsFor(topicName);
-
-								// do a sanity check on the partition set
-								if (partitions.size() < partitionCount) {
-									throw new IllegalStateException("The number of expected partitions was: "
-											+ partitionCount + ", but " + partitions.size()
-											+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
-								}
-								return partitions;
-							}
-						});
-			}
-			catch (Exception e) {
-				this.logger.error("Cannot initialize Binder", e);
-				throw new BinderException("Cannot initialize binder:", e);
-			}
-
 		}
 		finally {
 			zkUtils.close();
 		}
 	}

+	private Collection<PartitionInfo> getPartitionsForTopic(final String topicName, final int partitionCount) {
+		try {
+			return this.metadataRetryOperations
+					.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
+
+						@Override
+						public Collection<PartitionInfo> doWithRetry(RetryContext context) throws Exception {
+							Collection<PartitionInfo> partitions =
+									getProducerFactory(
+											new ExtendedProducerProperties<>(new KafkaProducerProperties()))
+											.createProducer().partitionsFor(topicName);
+
+							// do a sanity check on the partition set
+							if (partitions.size() < partitionCount) {
+								throw new IllegalStateException("The number of expected partitions was: "
+										+ partitionCount + ", but " + partitions.size()
+										+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
+							}
+							return partitions;
+						}
+					});
+		}
+		catch (Exception e) {
+			this.logger.error("Cannot initialize Binder", e);
+			throw new BinderException("Cannot initialize binder:", e);
+		}
+	}
+
 	private synchronized void initDlqProducer() {
 		try {
 			if (this.dlqProducer == null) {
@@ -549,8 +557,8 @@ public class KafkaMessageChannelBinder extends
 		private final DefaultKafkaProducerFactory<byte[], byte[]> producerFactory;

 		private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
-				ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
-				DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
+													ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
+													DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
 			super(kafkaTemplate);
 			setTopicExpression(new LiteralExpression(topic));
 			setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.admin;
+
+import java.util.Properties;
+
+import kafka.utils.ZkUtils;
+
+/**
+ * API around {@link kafka.admin.AdminUtils} to support
+ * various versions of Kafka brokers.
+ *
+ * Note: Implementations that support Kafka brokers other than 0.9, need to use
+ * a possible strategy that involves reflection around {@link kafka.admin.AdminUtils}.
+ *
+ * @author Soby Chacko
+ */
+public interface AdminUtilsOperation {
+
+	/**
+	 * Invoke {@link kafka.admin.AdminUtils#addPartitions}
+	 *
+	 * @param zkUtils Zookeeper utils
+	 * @param topic name of the topic
+	 * @param numPartitions
+	 * @param replicaAssignmentStr
+	 * @param checkBrokerAvailable
+	 */
+	void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions,
+							String replicaAssignmentStr, boolean checkBrokerAvailable);
+
+	/**
+	 * Invoke {@link kafka.admin.AdminUtils#fetchTopicMetadataFromZk}
+	 *
+	 * @param topic name
+	 * @param zkUtils zookeeper utils
+	 * @return error code
+	 */
+	short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils);
+
+	/**
+	 * Find partition size from Kafka broker using {@link kafka.admin.AdminUtils}
+	 *
+	 * @param topic name
+	 * @param zkUtils zookeeper utils
+	 * @return partition size
+	 */
+	int partitionSize(String topic, ZkUtils zkUtils);
+
+	/**
+	 * Inovke {@link kafka.admin.AdminUtils#createTopic}
+	 *
+	 * @param zkUtils zookeeper utils
+	 * @param topic name
+	 * @param partitions
+	 * @param replicationFactor
+	 * @param topicConfig
+	 */
+	void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions,
+					int replicationFactor, Properties topicConfig);
+}
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.admin;
+
+import java.util.Properties;
+
+import kafka.admin.AdminUtils;
+import kafka.api.TopicMetadata;
+import kafka.utils.ZkUtils;
+
+/**
+ * @author Soby Chacko
+ */
+public class Kafka09AdminUtilsOperation implements AdminUtilsOperation {
+
+	public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions,
+									String replicaAssignmentStr, boolean checkBrokerAvailable) {
+		AdminUtils.addPartitions(zkUtils, topic, numPartitions,
+				replicaAssignmentStr, checkBrokerAvailable);
+	}
+
+	public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) {
+		TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils);
+		return topicMetadata.errorCode();
+	}
+
+	public int partitionSize(String topic, ZkUtils zkUtils) {
+		TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils);
+		return topicMetadata.partitionsMetadata().size();
+	}
+
+	public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions,
+								int replicationFactor, Properties topicConfig) {
+		AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor,
+				topicConfig);
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java
@@ -0,0 +1,150 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.admin;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Properties;
+
+import kafka.api.PartitionMetadata;
+import kafka.utils.ZkUtils;
+
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * @author Soby Chacko
+ */
+public class Kafka10AdminUtilsOperation implements AdminUtilsOperation {
+
+	private static Class<?> ADMIN_UTIL_CLASS;
+
+	static {
+		try {
+			ADMIN_UTIL_CLASS = ClassUtils.forName("kafka.admin.AdminUtils", null);
+		}
+		catch (ClassNotFoundException e) {
+			throw new IllegalStateException("AdminUtils class not found", e);
+		}
+	}
+
+	public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions,
+									String replicaAssignmentStr, boolean checkBrokerAvailable) {
+		try {
+			Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods();
+			Method addPartitions = null;
+			for (Method m : declaredMethods) {
+				if (m.getName().equals("addPartitions")) {
+					addPartitions = m;
+				}
+			}
+
+			if (addPartitions != null) {
+				addPartitions.invoke(null, zkUtils, topic, numPartitions,
+						replicaAssignmentStr, checkBrokerAvailable, null);
+			}
+			else {
+				throw new InvocationTargetException(
+						new RuntimeException("method not found"));
+			}
+		}
+		catch (InvocationTargetException e) {
+			ReflectionUtils.handleInvocationTargetException(e);
+		}
+		catch (IllegalAccessException e) {
+			ReflectionUtils.handleReflectionException(e);
+		}
+	}
+
+	public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) {
+		try {
+			Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class);
+
+			Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils);
+			Class<?> topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null);
+
+			Method errorCodeMethod = ReflectionUtils.findMethod(topicMetadataClass, "error");
+			Object obj = errorCodeMethod.invoke(result);
+			Method code = ReflectionUtils.findMethod(obj.getClass(), "code");
+
+			return (short) code.invoke(obj);
+		}
+		catch (ClassNotFoundException e) {
+			throw new IllegalStateException("AdminUtils class not found", e);
+		}
+		catch (InvocationTargetException e) {
+			ReflectionUtils.handleInvocationTargetException(e);
+		}
+		catch (IllegalAccessException e) {
+			ReflectionUtils.handleReflectionException(e);
+		}
+		return 0;
+
+	}
+
+	@SuppressWarnings("unchecked")
+	public int partitionSize(String topic, ZkUtils zkUtils) {
+		try {
+			Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class);
+			Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils);
+			Class<?> topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null);
+
+			Method partitionsMetadata = ReflectionUtils.findMethod(topicMetadataClass, "partitionMetadata");
+			List<PartitionMetadata> foo = (List<PartitionMetadata>) partitionsMetadata.invoke(result);
+			return foo.size();
+		}
+		catch (ClassNotFoundException e) {
+			throw new IllegalStateException("AdminUtils class not found", e);
+		}
+		catch (InvocationTargetException e) {
+			ReflectionUtils.handleInvocationTargetException(e);
+		}
+		catch (IllegalAccessException e) {
+			ReflectionUtils.handleReflectionException(e);
+		}
+		return 0;
+	}
+
+	public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions,
+									int replicationFactor, Properties topicConfig) {
+		try {
+			Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods();
+			Method createTopic = null;
+			for (Method m : declaredMethods) {
+				if (m.getName().equals("createTopic") && m.getParameterTypes()[m.getParameterTypes().length - 1].getName().endsWith("RackAwareMode")) {
+					createTopic = m;
+					break;
+				}
+			}
+			if (createTopic != null) {
+				createTopic.invoke(null, zkUtils, topic, partitions,
+						replicationFactor, topicConfig, null);
+			}
+			else {
+				throw new InvocationTargetException(
+						new RuntimeException("method not found"));
+			}
+		}
+		catch (InvocationTargetException e) {
+			ReflectionUtils.handleInvocationTargetException(e);
+		}
+		catch (IllegalAccessException e) {
+			ReflectionUtils.handleReflectionException(e);
+		}
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.config;
+
+import java.util.HashMap;
+import java.util.Map;
+import javax.security.auth.login.AppConfigurationEntry;
+
+import org.springframework.util.Assert;
+
+/**
+ * Contains properties for setting up an {@link AppConfigurationEntry} that can be used
+ * for the Kafka or Zookeeper client.
+ *
+ * @author Marius Bogoevici
+ */
+public class JaasLoginModuleConfiguration {
+
+	private String loginModule = "com.sun.security.auth.module.Krb5LoginModule";
+
+	private AppConfigurationEntry.LoginModuleControlFlag controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED;
+
+	private Map<String,String> options = new HashMap<>();
+
+	public String getLoginModule() {
+		return loginModule;
+	}
+
+	public void setLoginModule(String loginModule) {
+		Assert.notNull(loginModule, "cannot be null");
+		this.loginModule = loginModule;
+	}
+
+	public String getControlFlag() {
+		return controlFlag.toString();
+	}
+
+	public AppConfigurationEntry.LoginModuleControlFlag getControlFlagValue() {
+		return controlFlag;
+	}
+
+	public void setControlFlag(String controlFlag) {
+		Assert.notNull(controlFlag, "cannot be null");
+		if (AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL.equals(controlFlag)) {
+			this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL;
+		}
+		else if (AppConfigurationEntry.LoginModuleControlFlag.REQUIRED.equals(controlFlag)) {
+			this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED;
+		}
+		else if (AppConfigurationEntry.LoginModuleControlFlag.REQUISITE.equals(controlFlag)) {
+			this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUISITE;
+		}
+		else if (AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT.equals(controlFlag)) {
+			this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT;
+		}
+		else {
+			throw new IllegalArgumentException(controlFlag + " is not a supported control flag");
+		}
+	}
+
+	public Map<String, String> getOptions() {
+		return options;
+	}
+
+	public void setOptions(Map<String, String> options) {
+		this.options = options;
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
@@ -16,18 +16,35 @@

 package org.springframework.cloud.stream.binder.kafka.config;

+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.common.utils.AppInfoParser;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
+import org.springframework.cloud.stream.binder.kafka.KafkaBinderJaasInitializerListener;
 import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
+import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation;
 import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationListener;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.core.type.AnnotatedTypeMetadata;
 import org.springframework.integration.codec.Codec;
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
@@ -45,6 +62,8 @@ import org.springframework.kafka.support.ProducerListener;
@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class})
 public class KafkaBinderConfiguration {

+	protected static final Log logger = LogFactory.getLog(KafkaBinderConfiguration.class);
+
 	@Autowired
 	private Codec codec;

@@ -57,13 +76,20 @@ public class KafkaBinderConfiguration {
 	@Autowired
 	private ProducerListener producerListener;

+	@Autowired
+	private ApplicationContext context;
+
+	@Autowired (required = false)
+	private AdminUtilsOperation adminUtilsOperation;
+
 	@Bean
 	KafkaMessageChannelBinder kafkaMessageChannelBinder() {
 		KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
 				this.configurationProperties);
 		kafkaMessageChannelBinder.setCodec(this.codec);
-		//kafkaMessageChannelBinder.setProducerListener(producerListener);
+		kafkaMessageChannelBinder.setProducerListener(producerListener);
 		kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties);
+		kafkaMessageChannelBinder.setAdminUtilsOperation(adminUtilsOperation);
 		return kafkaMessageChannelBinder;
 	}

@@ -77,4 +103,48 @@ public class KafkaBinderConfiguration {
 	KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
 		return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, this.configurationProperties);
 	}
+
+	@Bean(name = "adminUtilsOperation")
+	@Conditional(Kafka09Present.class)
+	@ConditionalOnClass(name = "kafka.admin.AdminUtils")
+	public AdminUtilsOperation kafka09AdminUtilsOperation() {
+		logger.info("AdminUtils selected: Kafka 0.9 AdminUtils");
+		return new Kafka09AdminUtilsOperation();
+	}
+
+	@Bean(name = "adminUtilsOperation")
+	@Conditional(Kafka10Present.class)
+	@ConditionalOnClass(name = "kafka.admin.AdminUtils")
+	public AdminUtilsOperation kafka10AdminUtilsOperation() {
+		logger.info("AdminUtils selected: Kafka 0.10 AdminUtils");
+		return new Kafka10AdminUtilsOperation();
+	}
+
+	@Bean
+	public ApplicationListener<?> jaasInitializer() throws IOException {
+		return new KafkaBinderJaasInitializerListener();
+	}
+
+	static class Kafka10Present implements Condition {
+
+		@Override
+		public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
+			return AppInfoParser.getVersion().startsWith("0.10");
+		}
+	}
+	
+	static class Kafka09Present implements Condition {
+
+		@Override
+		public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
+			return AppInfoParser.getVersion().startsWith("0.9");
+		}
+	}
+
+	public static class JaasConfigurationProperties {
+
+		private JaasLoginModuleConfiguration kafka;
+
+		private JaasLoginModuleConfiguration zookeeper;
+	}
 }
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java
@@ -77,6 +77,8 @@ public class KafkaBinderConfigurationProperties {

 	private int queueSize = 8192;

+	private JaasLoginModuleConfiguration jaas;
+
 	public String getZkConnectionString() {
 		return toConnectionString(this.zkNodes, this.defaultZkPort);
 	}
@@ -254,4 +256,13 @@ public class KafkaBinderConfigurationProperties {
 	public void setConfiguration(Map<String, String> configuration) {
 		this.configuration = configuration;
 	}
+
+	public JaasLoginModuleConfiguration getJaas() {
+		return jaas;
+	}
+
+	public void setJaas(JaasLoginModuleConfiguration jaas) {
+		this.jaas = jaas;
+	}
+
 }
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/package-info.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/package-info.java
@@ -1,20 +0,0 @@
-/*
- * Copyright 2015 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * This package contains an implementation of the {@link org.springframework.cloud.stream.binder.Binder} for Kafka.
- */
-package org.springframework.cloud.stream.binder.kafka;
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2014-2016 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.
@@ -13,7 +13,6 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-
 package org.springframework.cloud.stream.binder.kafka;

 import java.util.List;
@@ -24,49 +23,23 @@ import com.esotericsoftware.kryo.Registration;
 import org.springframework.cloud.stream.binder.AbstractTestBinder;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
-import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
-import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.codec.Codec;
 import org.springframework.integration.codec.kryo.KryoRegistrar;
 import org.springframework.integration.codec.kryo.PojoCodec;
 import org.springframework.integration.tuple.TupleKryoRegistrar;
-import org.springframework.kafka.support.LoggingProducerListener;
-import org.springframework.kafka.support.ProducerListener;

 /**
- * Test support class for {@link KafkaMessageChannelBinder}.
- * @author Eric Bottard
- * @author Marius Bogoevici
- * @author David Turanski
- * @author Gary Russell
 * @author Soby Chacko
 */
-public class KafkaTestBinder extends
+public abstract class AbstractKafkaTestBinder extends
 		AbstractTestBinder<KafkaMessageChannelBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {

-	public KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration) {
-		try {
-			KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration);
-			binder.setCodec(getCodec());
-			ProducerListener producerListener = new LoggingProducerListener();
-			binder.setProducerListener(producerListener);
-			GenericApplicationContext context = new GenericApplicationContext();
-			context.refresh();
-			binder.setApplicationContext(context);
-			binder.afterPropertiesSet();
-			this.setBinder(binder);
-		}
-		catch (Exception e) {
-			throw new RuntimeException(e);
-		}
-	}
-
 	@Override
 	public void cleanup() {
 		// do nothing - the rule will take care of that
 	}

-	private static Codec getCodec() {
+	protected static Codec getCodec() {
 		return new PojoCodec(new TupleRegistrar());
 	}

@@ -85,3 +58,4 @@ public class KafkaTestBinder extends
 	}

 }
+
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java
@@ -0,0 +1,164 @@
+/*
+ * Copyright 2014-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+import org.I0Itec.zkclient.ZkClient;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.Spy;
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.core.BrokerAddress;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.retry.RetryOperations;
+
+/**
+ * Integration tests for the {@link KafkaMessageChannelBinder}.
+ *
+ * @author Eric Bottard
+ * @author Marius Bogoevici
+ * @author Mark Fisher
+ * @author Ilayaperumal Gopinathan
+ */
+public class Kafka09BinderTests extends KafkaBinderTests {
+
+	private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10);
+
+	private Kafka09TestBinder binder;
+
+	private Kafka09AdminUtilsOperation adminUtilsOperation = new Kafka09AdminUtilsOperation();
+
+	@Override
+	protected void binderBindUnbindLatency() throws InterruptedException {
+		Thread.sleep(500);
+	}
+
+	@Override
+	protected Kafka09TestBinder getBinder() {
+		if (binder == null) {
+			KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
+			binder = new Kafka09TestBinder(binderConfiguration);
+		}
+		return binder;
+	}
+
+	protected KafkaBinderConfigurationProperties createConfigurationProperties() {
+		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
+		BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
+		List<String> bAddresses = new ArrayList<>();
+		for (BrokerAddress bAddress : brokerAddresses) {
+			bAddresses.add(bAddress.toString());
+		}
+		String[] foo = new String[bAddresses.size()];
+		binderConfiguration.setBrokers(bAddresses.toArray(foo));
+		binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString());
+		return binderConfiguration;
+	}
+
+	@Override
+	protected int partitionSize(String topic) {
+		return consumerFactory().createConsumer().partitionsFor(topic).size();
+	}
+
+	@Override
+	protected void setMetadataRetryOperations(Binder binder, RetryOperations retryOperations) {
+		((Kafka09TestBinder) binder).getBinder().setMetadataRetryOperations(retryOperations);
+	}
+
+	@Override
+	protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
+		final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(),
+				kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+
+		return new ZkUtils(zkClient, null, false);
+	}
+
+	@Override
+	protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) {
+		adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties());
+	}
+
+	@Override
+	protected int invokePartitionSize(String topic, ZkUtils zkUtils) {
+		return adminUtilsOperation.partitionSize(topic, zkUtils);
+	}
+
+	@Override
+	public String getKafkaOffsetHeaderKey() {
+		return KafkaHeaders.OFFSET;
+	}
+
+	@Override
+	protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
+		return new Kafka09TestBinder(kafkaBinderConfigurationProperties);
+	}
+
+	@Before
+	public void init() {
+		String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER");
+		if (multiplier != null) {
+			timeoutMultiplier = Double.parseDouble(multiplier);
+		}
+	}
+
+	@Override
+	protected boolean usesExplicitRouting() {
+		return false;
+	}
+
+	@Override
+	protected String getClassUnderTestName() {
+		return CLASS_UNDER_TEST_NAME;
+	}
+
+	@Override
+	public Spy spyOn(final String name) {
+		throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests");
+	}
+
+	private ConsumerFactory<byte[], byte[]> consumerFactory() {
+		Map<String, Object> props = new HashMap<>();
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+		Deserializer<byte[]> valueDecoder = new ByteArrayDeserializer();
+		Deserializer<byte[]> keyDecoder = new ByteArrayDeserializer();
+
+		return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder);
+	}
+
+}
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2015-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.kafka.support.LoggingProducerListener;
+import org.springframework.kafka.support.ProducerListener;
+
+/**
+ * Test support class for {@link KafkaMessageChannelBinder}. Creates a binder that uses
+ * an embedded Kafka cluster.
+ * @author Eric Bottard
+ * @author Marius Bogoevici
+ * @author David Turanski
+ * @author Gary Russell
+ * @author Soby Chacko
+ */
+public class Kafka09TestBinder extends AbstractKafkaTestBinder {
+
+	public Kafka09TestBinder(KafkaBinderConfigurationProperties binderConfiguration) {
+		try {
+			KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration);
+			binder.setCodec(getCodec());
+			ProducerListener producerListener = new LoggingProducerListener();
+			binder.setProducerListener(producerListener);
+			GenericApplicationContext context = new GenericApplicationContext();
+			context.refresh();
+			binder.setApplicationContext(context);
+			binder.setAdminUtilsOperation(new Kafka09AdminUtilsOperation());
+			binder.afterPropertiesSet();
+			this.setBinder(binder);
+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+}
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.lang.reflect.Field;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
+import org.springframework.kafka.support.ProducerListener;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = KafkaBinderConfiguration.class)
+public class KafkaBinderConfigurationTest {
+
+	@Autowired
+	private KafkaMessageChannelBinder kafkaMessageChannelBinder;
+
+	@Test
+	public void testKafkaBinderProducerListener() {
+		assertNotNull(this.kafkaMessageChannelBinder);
+		Field producerListenerField = ReflectionUtils.findField(
+				KafkaMessageChannelBinder.class, "producerListener",
+				ProducerListener.class);
+		ReflectionUtils.makeAccessible(producerListenerField);
+		ProducerListener producerListener = (ProducerListener) ReflectionUtils.getField(
+				producerListenerField, this.kafkaMessageChannelBinder);
+		assertNotNull(producerListener);
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import javax.security.auth.login.AppConfigurationEntry;
+
+import com.sun.security.auth.login.ConfigFile;
+import org.apache.kafka.common.security.JaasUtils;
+import org.junit.Test;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.ClassPathResource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KafkaBinderJaasInitializerListenerTest {
+
+	@Test
+	public void testConfigurationParsedCorrectlyWithKafkaClient() throws Exception {
+		ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
+		final AppConfigurationEntry[] kafkaConfigurationArray = configFile.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT);
+
+		final ConfigurableApplicationContext context =
+				SpringApplication.run(SimpleApplication.class,
+						"--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true",
+						"--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true",
+						"--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab",
+						"--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM");
+		javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration.getConfiguration();
+
+		final AppConfigurationEntry[] kafkaConfiguration = configuration.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT);
+		assertThat(kafkaConfiguration).hasSize(1);
+		assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions());
+		context.close();
+	}
+
+	@SpringBootApplication
+	public static class SimpleApplication {
+
+	}
+}
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java
@@ -40,12 +40,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 * @author Gary Russell
 * @author Mark Fisher
 */
-public class RawModeKafkaBinderTests extends KafkaBinderTests {
+public class RawModeKafka09BinderTests extends Kafka09BinderTests {

 	@Test
 	@Override
 	public void testPartitionedModuleJava() throws Exception {
-		KafkaTestBinder binder = getBinder();
+		Kafka09TestBinder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setHeaderMode(HeaderMode.raw);
 		properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class);
@@ -98,7 +98,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 	@Test
 	@Override
 	public void testPartitionedModuleSpEL() throws Exception {
-		KafkaTestBinder binder = getBinder();
+		Kafka09TestBinder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]"));
 		properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
@@ -160,7 +160,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 	@Test
 	@Override
 	public void testSendAndReceive() throws Exception {
-		KafkaTestBinder binder = getBinder();
+		Kafka09TestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		QueueChannel moduleInputChannel = new QueueChannel();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -184,7 +184,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {

 	@Test
 	public void testSendAndReceiveWithExplicitConsumerGroup() {
-		KafkaTestBinder binder = getBinder();
+		Kafka09TestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		// Test pub/sub by emulating how StreamPlugin handles taps
 		QueueChannel module1InputChannel = new QueueChannel();
--- a/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf
+++ b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf
@@ -0,0 +1,7 @@
+KafkaClient {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/security/keytabs/kafka_client.keytab"
+    principal="kafka-client-1@EXAMPLE.COM";
+};
--- a/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf
+++ b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf
@@ -0,0 +1,14 @@
+KafkaClient {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/security/keytabs/kafka_client.keytab"
+    principal="kafka-client-1@EXAMPLE.COM";
+};
+Client {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/security/keytabs/zk_client.keytab"
+    principal="zk-client-1@EXAMPLE.COM";
+};
--- a/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml
+++ b/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml
@@ -0,0 +1,14 @@
+<configuration>
+    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
+        </encoder>
+    </appender>
+    <logger name="org.springframework.integration.kafka" level="INFO"/>
+    <logger name="org.springframework.kafka" level="INFO"/>
+    <logger name="org.springframework.cloud.stream" level="INFO" />
+    <logger name="org.springframework.integration.channel" level="INFO" />
+    <root level="WARN">
+        <appender-ref ref="stdout"/>
+    </root>
+</configuration>
Author	SHA1	Message	Date
Marius Bogoevici	b29498617d	Release 1.1.1.RELEASE Signed-off-by: Marius Bogoevici <mbogoevici@pivotal.io>	2017-01-11 22:32:42 -05:00
Marius Bogoevici	547ea77b2f	Update mvnw	2017-01-11 22:13:52 -05:00
Marius Bogoevici	699dc06bc5	Observe `bufferSize` setting Fixes #77 Signed-off-by: Marius Bogoevici <mbogoevici@pivotal.io>	2017-01-11 22:13:29 -05:00
Marius Bogoevici	e822e59859	Remove package-info.java	2017-01-11 22:07:38 -05:00
Marius Bogoevici	cc630f2be0	Remove explicit version references for Stream and Kafka binder artifacts Signed-off-by: Marius Bogoevici <mbogoevici@pivotal.io>	2017-01-04 22:33:38 -05:00
Marius Bogoevici	1668f0c694	Update to using 'lifecycle' instead of 'endpoint' for tests	2016-12-16 18:13:33 -05:00
Ilayaperumal Gopinathan	36c90aa8c0	Ability to override deserializer in Kafka consumer - When binding the consumer, the kafka consumer should not be set to use `ByteArrayDeserializer` for both key/value deserializer. Instead, they need to be used as the default values. Any extended consumer properties for key/value deserializer should override this default deserializer. - Add test This resolves #55 Add tests for custom/native serialization - Test using built-in serialization without using kafka native serialization (ie. both the serializer/de-serializer are set to use ByteArraySe/Deserializer) - Test using custom serializer by explicitly setting value.deserializer for both Kafka producer/consumer properties - Test avro message conversion and Kafka Avro Serializer using Confluent Schema Registry - Given pre-released registry versions have some bugs that blocks the testing and `3.0.1` requires Kafka 0.10, this test is only in Kafka .10 binder Update Schema based custom serializer/de-serializer tests Fix import Handle unbind during tests appropriately	2016-11-15 18:20:34 -05:00
Artem Bilan	943a7ae148	Upgrade to Spring Kafka 1.0.5 Also fix `mvnw.cmd` to be really for Windows	2016-11-08 11:34:34 -05:00
Soby Chacko	a1633ab241	Adding latency after unbind in testResume to ensure offsets are updated in broker	2016-10-24 21:34:56 -04:00
Soby Chacko	68fd3fb6d7	enable auto rebalance in testResume	2016-10-24 11:05:54 -04:00
Soby Chacko	df6d472e99	Enable auto rebalancing in testCompression Reduce logging levels in tests	2016-10-21 15:21:09 -04:00
Soby Chacko	23ae2b3088	setting producer to publish synchronously in tests	2016-10-18 22:50:36 -04:00
Soby Chacko	2e595f34f2	Increase logging in tests	2016-10-18 22:48:55 -04:00
Soby Chacko	acb1c60216	test cleanup	2016-10-18 19:45:44 -04:00
Marius Bogoevici	275f5cbb8e	Use Spring Cloud Build as parent Fixes #60	2016-10-17 20:10:06 +05:30
Soby Chacko	79643586dd	Ensure that topic is created where broker automatically creates it	2016-10-14 17:31:18 -04:00
Ilayaperumal Gopinathan	b292f81d46	Set producer listener obtained from the context - When creating the KafkaMessageChannelBinder set the producer listener that is being autowired in the configuration - Add test This resolves #49	2016-10-04 19:03:40 +05:30
Marius Bogoevici	8ad98c09b9	Revert to Spring Cloud Stream 1.1.1.BUILD-SNAPSHOT	2016-09-22 15:29:58 -04:00
bamboo	624b039da5	[artifactory-release] Next development version	2016-09-22 01:26:20 +00:00
bamboo	b08888862a	[artifactory-release] Release version 1.1.0.RELEASE	2016-09-22 01:26:20 +00:00
Marius Bogoevici	e13e1b1b1b	Fix Kafka tests As ConsumerProperties are initialized with different defaults tests must configure them with the same logic as ChannelBindingServiceProperties.	2016-09-21 20:32:36 -04:00
Marius Bogoevici	513c27beb8	Update to Spring Cloud Stream 1.1.0.RELEASE	2016-09-21 19:01:19 -04:00
Soby Chacko	5f1e6caeab	Manual acking tests Adding tests to verify manual acking when auto commit offsets are disabled/enabled. Docs changes.	2016-09-21 18:52:27 -04:00
Soby Chacko	ed4416e7c7	Upgrading spring-kafka and spring-integration-kafka dependencies, docs changes	2016-09-21 18:10:46 -04:00
Nadelson, Mark	546fe69346	set Kafka ackMode to MANUAL if autoCommitOffset = false polishing	2016-09-21 18:04:02 -04:00
Marius Bogoevici	30c868e51b	Configure JAAS settings for Kafka via Spring Boot Fixes #38 Adds support for programmatic configurtion of JAAS properties as an alternative to using a JAAS file. Add a placeholder JAAS configuration file which is ignored by the configuration but lets Kafka 0.9 clients connect	2016-09-20 11:05:21 -04:00
Marius Bogoevici	6506018a2c	Update README.adoc	2016-09-20 09:15:17 -04:00
Marius Bogoevici	09ec23b4bc	Minor polishing and refactoring	2016-09-14 15:24:38 -04:00
Soby Chacko	8f4fbd0da6	Making AdminUtils optional polishing, docs Addressing PR review comments Polishing: Refactoring topic creation methods Addressing PR comments: Use ClassUtils, Kafka AppInfoParser and polishing	2016-09-13 17:34:57 -04:00
bamboo	323206130a	[artifactory-release] Next development version	2016-09-07 23:38:15 +00:00
bamboo	b944b594b5	[artifactory-release] Release version 1.1.0.RC2	2016-09-07 23:38:15 +00:00
Marius Bogoevici	4fa567fec7	Fix documentation issue	2016-09-07 19:14:31 -04:00
bamboo	16d6584774	[artifactory-release] Next development version	2016-09-07 22:38:01 +00:00
bamboo	ea514bd72d	[artifactory-release] Release version 1.1.0.RC1	2016-09-07 22:38:01 +00:00
Marius Bogoevici	79feac15d0	Upgrade to Spring Cloud Stream 1.1.0.RC1	2016-09-07 18:25:49 -04:00
Marius Bogoevici	3a27a5ec75	Remove redundant spring-kafka declaration	2016-09-07 18:25:22 -04:00
Marius Bogoevici	7308bd4991	Polishing Removed optional flag	2016-09-07 17:53:49 -04:00
Soby Chacko	866eaf4a25	Add support for drop-in support for Kafka 0.10 Reflectively detect AdminUtils from Kafka 0.9 and 0.10 Introduce Kafka 10 conditionals	2016-09-07 17:52:53 -04:00
bamboo	b37e7b37d2	[artifactory-release] Next development version	2016-08-26 03:30:26 +00:00