Initial Commit

Root pom.xml - spring-cloud-stream-acceptance-tests-parent
Local acceptance tests based on spring-cloud-stream-samples

.gitignore

@@ -0,0 +1,27 @@
+/application.yml
+/application.properties
+asciidoctor.css
+*~
+.#*
+*#
+target/
+build/
+bin/
+_site/
+.classpath
+.project
+.settings
+.springBeans
+.sts4-cache/
+.attach_pid*
+.DS_Store
+*.sw*
+*.iml
+*.ipr
+*.iws
+.idea/*
+.factorypath
+dump.rdb
+.apt_generated
+artifacts
+**/dependency-reduced-pom.xml

.mvn/jvm.config

@@ -0,0 +1 @@
+-Xmx1024m -XX:CICompilerCount=1 -XX:TieredStopAtLevel=1 -Djava.security.egd=file:/dev/./urandom

.mvn/maven.config

@@ -0,0 +1 @@
+-DaltSnapshotDeploymentRepository=repo.spring.io::default::https://repo.spring.io/libs-snapshot-local -P spring

.mvn/wrapper/maven-wrapper.properties

@@ -0,0 +1 @@
+distributionUrl=https://repo1.maven.org/maven2/org/apache/maven/apache-maven/3.3.9/apache-maven-3.3.9-bin.zip

pom.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>spring-cloud-stream-acceptance-tests-parent</artifactId>
+    <groupId>io.spring.cloud.stream.acceptance</groupId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>pom</packaging>
+    <name>spring-cloud-stream-acceptance-tests-parent</name>
+    <description>Spring Cloud Stream Acceptnace Tests Parent</description>
+
+    <parent>
+        <groupId>org.springframework.cloud</groupId>
+        <artifactId>spring-cloud-build</artifactId>
+        <version>2.1.0.BUILD-SNAPSHOT</version>
+    </parent>
+
+    <modules>
+      <module>spring-cloud-stream-local-acceptance-tests</module>
+      <!--<module>spring-cloud-stream-cf-acceptance-tests</module>-->
+      <!--<module>spring-cloud-stream-k8s-acceptance-tests</module>-->
+    </modules>
+</project>
+

spring-cloud-stream-local-acceptance-tests/.mvn

@@ -0,0 +1 @@
+../.mvn

spring-cloud-stream-local-acceptance-tests/README.adoc

@@ -0,0 +1,10 @@
+=== Samples Acceptance Tests
+
+This is an accptance test module for the samples in this repo.
+The tests launch the Spring Cloud Stream samples as stand alone Spring Boot applications and then verify their correctness.
+
+By default, these tests are not run as part of the normal build, as they are mainly intended for continuous integration testing with ongoing changes in the framework.
+
+In order to run the tests, we recommend to run the script `./runAcceptanceTest.sh` in this directory.
+The script will launch all the middleware and other components in docker containers first.
+Then it builds the applications and run them.

spring-cloud-stream-local-acceptance-tests/docker-compose.yml

@@ -0,0 +1,54 @@
+version: '2'
+volumes:
+  data-volume: {}
+services:
+  mysql:
+    image: mariadb
+    ports:
+      - "3306:3306"
+    environment:
+      MYSQL_ROOT_PASSWORD: pwd
+      MYSQL_DATABASE: sample_mysql_db
+    volumes:
+      - data-volume:/var/lib/mysql
+  kafka:
+    image: wurstmeister/kafka
+    ports:
+      - "9092:9092"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=127.0.0.1
+      - KAFKA_ADVERTISED_PORT=9092
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+    depends_on:
+      - zookeeper
+  zookeeper:
+    image: wurstmeister/zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=zookeeper
+  rabbitmq:
+    image: rabbitmq:management
+    ports:
+      - 5672:5672
+      - 15672:15672
+
+  # used for multi Kafka cluster testing
+
+  kafka2:
+    image: wurstmeister/kafka
+    container_name: kafka-2
+    ports:
+      - "9093:9092"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=127.0.0.1
+      - KAFKA_ADVERTISED_PORT=9092
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper2:2181
+    depends_on:
+      - zookeeper2
+  zookeeper2:
+    image: wurstmeister/zookeeper
+    ports:
+      - "2182:2181"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=zookeeper2

spring-cloud-stream-local-acceptance-tests/mvnw

@@ -0,0 +1,226 @@
+#!/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.
+# ----------------------------------------------------------------------------
+
+# ----------------------------------------------------------------------------
+# 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
+# ----------------------------------------------------------------------------
+
+if [ -z "$MAVEN_SKIP_RC" ] ; then
+
+  if [ -f /etc/mavenrc ] ; then
+    . /etc/mavenrc
+  fi
+
+  if [ -f "$HOME/.mavenrc" ] ; then
+    . "$HOME/.mavenrc"
+  fi
+
+fi
+
+# 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
+    # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
+    # See https://developer.apple.com/library/mac/qa/qa1170/_index.html
+    if [ -z "$JAVA_HOME" ]; then
+      if [ -x "/usr/libexec/java_home" ]; then
+        export JAVA_HOME="`/usr/libexec/java_home`"
+      else
+        export JAVA_HOME="/Library/Java/Home"
+      fi
+    fi
+    ;;
+esac
+
+if [ -z "$JAVA_HOME" ] ; then
+  if [ -r /etc/gentoo-release ] ; then
+    JAVA_HOME=`java-config --jre-home`
+  fi
+fi
+
+if [ -z "$M2_HOME" ] ; then
+  ## resolve links - $0 may be a link to maven's home
+  PRG="$0"
+
+  # 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
+
+  saveddir=`pwd`
+
+  M2_HOME=`dirname "$PRG"`/..
+
+  # make it fully qualified
+  M2_HOME=`cd "$M2_HOME" && pwd`
+
+  cd "$saveddir"
+  # echo Using m2 at $M2_HOME
+fi
+
+# 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
+
+# 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
+
+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
+
+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
+
+if [ ! -x "$JAVACMD" ] ; then
+  echo "Error: JAVA_HOME is not defined correctly." >&2
+  echo "  We cannot execute $JAVACMD" >&2
+  exit 1
+fi
+
+if [ -z "$JAVA_HOME" ] ; then
+  echo "Warning: JAVA_HOME environment variable is not set."
+fi
+
+CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
+
+# traverses directory structure from process work directory to filesystem root
+# first directory with .mvn subdirectory is considered project base directory
+find_maven_basedir() {
+
+  if [ -z "$1" ]
+  then
+    echo "Path not specified to find_maven_basedir"
+    return 1
+  fi
+
+  basedir="$1"
+  wdir="$1"
+  while [ "$wdir" != '/' ] ; do
+    if [ -d "$wdir"/.mvn ] ; then
+      basedir=$wdir
+      break
+    fi
+    # workaround for JBEAP-8937 (on Solaris 10/Sparc)
+    if [ -d "${wdir}" ]; then
+      wdir=`cd "$wdir/.."; pwd`
+    fi
+    # end of workaround
+  done
+  echo "${basedir}"
+}
+
+# concatenates all lines of a file
+concat_lines() {
+  if [ -f "$1" ]; then
+    echo "$(tr -s '\n' ' ' < "$1")"
+  fi
+}
+
+BASE_DIR=`find_maven_basedir "$(pwd)"`
+if [ -z "$BASE_DIR" ]; then
+  exit 1;
+fi
+
+export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
+echo $MAVEN_PROJECTBASEDIR
+MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
+
+# 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"`
+  [ -n "$MAVEN_PROJECTBASEDIR" ] &&
+    MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
+fi
+
+WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+"$JAVACMD" \
+  $MAVEN_OPTS \
+  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
+  "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
+  ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"
+

spring-cloud-stream-local-acceptance-tests/mvnw.cmd

@@ -0,0 +1,145 @@
+@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 ----------------------------------------------------------------------------
+
+@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 ----------------------------------------------------------------------------
+
+@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%
+
+@REM set %HOME% to equivalent of $HOME
+if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
+
+@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
+
+@setlocal
+
+set ERROR_CODE=0
+
+@REM To isolate internal variables from possible post scripts, we use another setlocal
+@setlocal
+
+@REM ==== START VALIDATION ====
+if not "%JAVA_HOME%" == "" goto OkJHome
+
+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
+
+:OkJHome
+if exist "%JAVA_HOME%\bin\java.exe" goto init
+
+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
+
+@REM ==== END VALIDATION ====
+
+:init
+
+set MAVEN_CMD_LINE_ARGS=%*
+
+@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
+@REM Fallback to current working directory if not found.
+
+set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
+IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
+
+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
+
+:baseDirFound
+set MAVEN_PROJECTBASEDIR=%WDIR%
+cd "%EXEC_DIR%"
+goto endDetectBaseDir
+
+:baseDirNotFound
+set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
+cd "%EXEC_DIR%"
+
+:endDetectBaseDir
+
+IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
+
+@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%
+
+:endReadAdditionalConfig
+
+SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
+
+set WRAPPER_JAR="".\.mvn\wrapper\maven-wrapper.jar""
+set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+%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
+
+:error
+set ERROR_CODE=1
+
+: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%

spring-cloud-stream-local-acceptance-tests/pom.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>spring-cloud-stream-local-acceptance-tests</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>jar</packaging>
+    <name>spring-cloud-stream-local-acceptance-tests</name>
+    <description>Spring Cloud Stream Local Acceptance Tests</description>
+
+    <parent>
+        <groupId>io.spring.cloud.stream.acceptance</groupId>
+        <artifactId>spring-cloud-stream-acceptance-tests-parent</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../</relativePath>
+    </parent>
+
+    <properties>
+        <maven.test.skip>true</maven.test.skip>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mariadb.jdbc</groupId>
+            <artifactId>mariadb-java-client</artifactId>
+            <version>1.1.9</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

spring-cloud-stream-local-acceptance-tests/runAcceptanceTests.sh

@@ -0,0 +1,136 @@
+
+#!/bin/bash
+
+pushd () {
+    command pushd "$@" > /dev/null
+}
+
+popd () {
+    command popd "$@" > /dev/null
+}
+
+function prepare_jdbc_source_with_kafka_and_rabbit_binders() {
+    wget -O /tmp/jdbc-source-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sample-jdbc-source/0.0.1-SNAPSHOT/sample-jdbc-source-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/jdbc-source-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sample-jdbc-source/0.0.1-SNAPSHOT/sample-jdbc-source-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_jdbc_sink_with_kafka_and_rabbit_binders() {
+    wget -O /tmp/jdbc-sink-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sample-jdbc-sink/0.0.1-SNAPSHOT/sample-jdbc-sink-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/jdbc-sink-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sample-jdbc-sink/0.0.1-SNAPSHOT/sample-jdbc-sink-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_dynamic_source_with_kafka_and_rabbit_binders() {
+    wget -O /tmp/dynamic-destination-source-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/dynamic-destination-source/0.0.1-SNAPSHOT/dynamic-destination-source-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/dynamic-destination-source-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/dynamic-destination-source/0.0.1-SNAPSHOT/dynamic-destination-source-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_multi_binder_with_kafka_rabbit() {
+    wget -O /tmp/multibinder-kafka-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/multibinder-kafka-rabbit/0.0.1-SNAPSHOT/multibinder-kafka-rabbit-0.0.1-SNAPSHOT.jar
+}
+
+function prepare_multi_binder_with_two_kafka_clusters() {
+    wget -O /tmp/multibinder-two-kafka-clusters-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/multibinder-two-kafka-clusters/0.0.1-SNAPSHOT/multibinder-two-kafka-clusters-0.0.1-SNAPSHOT.jar
+}
+
+function prepare_kafka_streams_word_count() {
+    wget -O /tmp/kafka-streams-word-count-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/kafka-streams-word-count/0.0.1-SNAPSHOT/kafka-streams-word-count-0.0.1-SNAPSHOT.jar
+}
+
+function prepare_streamlistener_basic_with_kafka_rabbit_binders() {
+    wget -O /tmp/streamlistener-basic-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/streamlistener-basic/0.0.1-SNAPSHOT/streamlistener-basic-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/streamlistener-basic-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/streamlistener-basic/0.0.1-SNAPSHOT/streamlistener-basic-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_reactive_processor_with_kafka_rabbit_binders() {
+    wget -O /tmp/reactive-processor-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/reactive-processor/0.0.1-SNAPSHOT/reactive-processor-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/reactive-processor-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/reactive-processor/0.0.1-SNAPSHOT/reactive-processor-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_sensor_average_reactive_with_kafka_rabbit_binders() {
+    wget -O /tmp/sensor-average-reactive-kafka-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sensor-average-reactive/0.0.1-SNAPSHOT/sensor-average-reactive-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/sensor-average-reactive-rabbit-sample.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/sensor-average-reactive/0.0.1-SNAPSHOT/sensor-average-reactive-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_schema_registry_vanilla_with_kafka_rabbit_binders() {
+    wget -O /tmp/schema-registry-vanilla-registry-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-server/0.0.1-SNAPSHOT/schema-registry-vanilla-server-0.0.1-SNAPSHOT.jar
+    wget -O /tmp/schema-registry-vanilla-consumer-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-consumer/0.0.1-SNAPSHOT/schema-registry-vanilla-consumer-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/schema-registry-vanilla-producer1-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-producer1/0.0.1-SNAPSHOT/schema-registry-vanilla-producer1-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/schema-registry-vanilla-producer2-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-producer2/0.0.1-SNAPSHOT/schema-registry-vanilla-producer2-0.0.1-SNAPSHOT-kafka.jar
+
+    wget -O /tmp/schema-registry-vanilla-registry-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-server/0.0.1-SNAPSHOT/schema-registry-vanilla-server-0.0.1-SNAPSHOT.jar
+    wget -O /tmp/schema-registry-vanilla-consumer-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-consumer/0.0.1-SNAPSHOT/schema-registry-vanilla-consumer-0.0.1-SNAPSHOT-rabbit.jar
+    wget -O /tmp/schema-registry-vanilla-producer1-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-producer1/0.0.1-SNAPSHOT/schema-registry-vanilla-producer1-0.0.1-SNAPSHOT-rabbit.jar
+    wget -O /tmp/schema-registry-vanilla-producer2-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/schema-registry-vanilla-producer2/0.0.1-SNAPSHOT/schema-registry-vanilla-producer2-0.0.1-SNAPSHOT-rabbit.jar
+}
+
+function prepare_partitioning_with_kafka_rabbit_binders() {
+    wget -O /tmp/partitioning-producer-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/partitioning-producer/0.0.1-SNAPSHOT/partitioning-producer-0.0.1-SNAPSHOT-kafka.jar
+    wget -O /tmp/partitioning-consumer-kafka.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/partitioning-consumer-kafka/0.0.1-SNAPSHOT/partitioning-consumer-kafka-0.0.1-SNAPSHOT.jar
+
+    wget -O /tmp/partitioning-producer-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/partitioning-producer/0.0.1-SNAPSHOT/partitioning-producer-0.0.1-SNAPSHOT-rabbit.jar
+    wget -O /tmp/partitioning-consumer-rabbit.jar http://repo.spring.io/libs-snapshot-local/io/spring/cloud/stream/sample/partitioning-consumer-rabbit/0.0.1-SNAPSHOT/partitioning-consumer-rabbit-0.0.1-SNAPSHOT.jar
+}
+
+#Main script starting
+
+echo "Prepare artifacts for testing"
+
+prepare_jdbc_source_with_kafka_and_rabbit_binders
+prepare_jdbc_sink_with_kafka_and_rabbit_binders
+prepare_dynamic_source_with_kafka_and_rabbit_binders
+prepare_multi_binder_with_kafka_rabbit
+prepare_multi_binder_with_two_kafka_clusters
+prepare_streamlistener_basic_with_kafka_rabbit_binders
+prepare_reactive_processor_with_kafka_rabbit_binders
+prepare_sensor_average_reactive_with_kafka_rabbit_binders
+prepare_kafka_streams_word_count
+
+prepare_schema_registry_vanilla_with_kafka_rabbit_binders
+
+prepare_partitioning_with_kafka_rabbit_binders
+
+echo "Starting components in docker containers..."
+
+docker-compose up -d
+
+echo "Running tests"
+
+./mvnw clean package -Dmaven.test.skip=false
+BUILD_RETURN_VALUE=$?
+
+docker-compose down
+
+# Post cleanup
+
+rm /tmp/jdbc-source-kafka-sample.jar
+rm /tmp/jdbc-source-rabbit-sample.jar
+rm /tmp/jdbc-sink-kafka-sample.jar
+rm /tmp/jdbc-sink-rabbit-sample.jar
+rm /tmp/dynamic-destination-source-kafka-sample.jar
+rm /tmp/dynamic-destination-source-rabbit-sample.jar
+rm /tmp/multibinder-kafka-rabbit-sample.jar
+rm /tmp/multibinder-two-kafka-clusters-sample.jar
+rm /tmp/kafka-streams-word-count-sample.jar
+rm /tmp/streamlistener-basic-kafka-sample.jar
+rm /tmp/streamlistener-basic-rabbit-sample.jar
+rm /tmp/reactive-processor-kafka-sample.jar
+rm /tmp/reactive-processor-rabbit-sample.jar
+rm /tmp/sensor-average-reactive-kafka-sample.jar
+rm /tmp/sensor-average-reactive-rabbit-sample.jar
+
+rm /tmp/schema-registry-vanilla-registry-kafka.jar
+rm /tmp/schema-registry-vanilla-consumer-kafka.jar
+rm /tmp/schema-registry-vanilla-producer1-kafka.jar
+rm /tmp/schema-registry-vanilla-producer2-kafka.jar
+rm /tmp/schema-registry-vanilla-registry-rabbit.jar
+rm /tmp/schema-registry-vanilla-consumer-rabbit.jar
+rm /tmp/schema-registry-vanilla-producer1-rabbit.jar
+rm /tmp/schema-registry-vanilla-producer2-rabbit.jar
+
+rm /tmp/partitioning-producer-kafka.jar
+rm /tmp/partitioning-consumer-kafka.jar
+
+rm /tmp/partitioning-producer-rabbit.jar
+rm /tmp/partitioning-consumer-rabbit.jar
+
+exit $BUILD_RETURN_VALUE

spring-cloud-stream-local-acceptance-tests/src/test/java/sample/acceptance/tests/AbstractSampleTests.java

@@ -0,0 +1,78 @@
+package sample.acceptance.tests;
+
+import org.assertj.core.util.Files;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.StringUtils;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.fail;
+
+/**
+ * @author Soby Chacko
+ */
+public abstract class AbstractSampleTests {
+
+	private static final Logger logger = LoggerFactory.getLogger(AbstractSampleTests.class);
+
+	protected boolean waitForLogEntryInFile(String app, File f, String... entries) {
+		logger.info("Looking for '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
+		long timeout = System.currentTimeMillis() + (60 * 1000);
+		boolean exists = false;
+		while (!exists && System.currentTimeMillis() < timeout) {
+			try {
+				Thread.sleep(2 * 1000);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new IllegalStateException(e.getMessage(), e);
+			}
+			logger.info("Polling to get log file. Remaining poll time = "
+					+ (timeout - System.currentTimeMillis() + " ms."));
+			String log = Files.contentOf(f, StandardCharsets.UTF_8);
+
+			if (log != null) {
+				if (Stream.of(entries).allMatch(log::contains)) {
+					exists = true;
+				}
+			}
+		}
+		if (exists) {
+			logger.info("Matched all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for app " + app);
+		} else {
+			logger.error("ERROR: Couldn't find all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
+			fail("Could not find the test data in the logs");
+		}
+		return true;
+	}
+
+	protected boolean waitForLogEntryInFileWithoutFailing(String app, File f, String... entries) {
+		logger.info("Looking for '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
+		long timeout = System.currentTimeMillis() + (60 * 1000);
+		boolean exists = false;
+		while (!exists && System.currentTimeMillis() < timeout) {
+			try {
+				Thread.sleep(2 * 1000);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new IllegalStateException(e.getMessage(), e);
+			}
+			logger.info("Polling to get log file. Remaining poll time = "
+					+ (timeout - System.currentTimeMillis() + " ms."));
+			String log = Files.contentOf(f, StandardCharsets.UTF_8);
+
+			if (log != null) {
+				if (Stream.of(entries).allMatch(log::contains)) {
+					exists = true;
+				}
+			}
+		}
+		if (exists) {
+			logger.info("Matched all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for app " + app);
+		}
+		return exists;
+	}
+
+}

spring-cloud-stream-local-acceptance-tests/src/test/java/sample/acceptance/tests/PartitioningAcceptanceTests.java

@@ -0,0 +1,201 @@
+package sample.acceptance.tests;
+
+import org.assertj.core.util.Files;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static org.junit.Assert.fail;
+
+/**
+ * @author Soby Chacko
+ */
+public class PartitioningAcceptanceTests extends AbstractSampleTests {
+
+	private static final Logger logger = LoggerFactory.getLogger(PartitioningAcceptanceTests.class);
+
+	@Test
+	public void testPartitioningKafka() throws Exception {
+		Process producerProcess = null;
+		Process consumer1Process = null;
+		Process consumer2Process = null;
+
+		try {
+			ProcessBuilder producerProcessBuilder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-producer-kafka.jar");
+			File producerFile = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + producerFile.getAbsolutePath());
+			producerProcessBuilder.redirectOutput(producerFile);
+			producerProcess = producerProcessBuilder.start();
+
+			waitForLogEntryInFile("Partitioning producer", producerFile, "Started PartProducerApplication in");
+
+			ProcessBuilder consumer1Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-kafka.jar", "--server.port=12001");
+			File consumer1File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer1File.getAbsolutePath());
+			consumer1Builder.redirectOutput(consumer1File);
+			consumer1Process = consumer1Builder.start();
+
+			ProcessBuilder consumer2Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-kafka.jar", "--server.port=12002");
+			File consumer2File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer2File.getAbsolutePath());
+			consumer2Builder.redirectOutput(consumer2File);
+			consumer2Process = consumer2Builder.start();
+
+			Future<?> future1 = verifyPartitions("Partitioning Consumer-1", consumer1File, "Partitioning Consumer-2", consumer2File,
+					"f received from partition 0", "g received from partition 0", "h received from partition 0");
+			Future<?> future2 = verifyPartitions("Partitioning Consumer-1", consumer1File, "Partitioning Consumer-2", consumer2File,
+					"fo received from partition 1", "go received from partition 1", "ho received from partition 1");
+			Future<?> future3 = verifyPartitions("Partitioning Consumer-2",consumer2File, "Partitioning Consumer-1", consumer1File,
+					"foo received from partition 2", "goo received from partition 2", "hoo received from partition 2");
+			Future<?> future4 = verifyPartitions("Partitioning Consumer-2",consumer2File, "Partitioning Consumer-1", consumer1File,
+					"fooz received from partition 3", "gooz received from partition 3", "hooz received from partition 3");
+
+			verifyResults(future1, future2, future3, future4);
+		}
+		finally {
+			if (producerProcess != null) {
+				producerProcess.destroyForcibly();
+			}
+			if (consumer1Process != null) {
+				consumer1Process.destroyForcibly();
+			}
+			if (consumer2Process != null) {
+				consumer2Process.destroyForcibly();
+			}
+		}
+	}
+
+	@Test
+	public void testPartitioningRabbit() throws Exception {
+		Process producerProcess = null;
+		Process consumer1Process = null;
+		Process consumer2Process = null;
+		Process consumer3Process = null;
+		Process consumer4Process = null;
+
+		try {
+			ProcessBuilder producerProcessBuilder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-producer-rabbit.jar");
+			File producerFile = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + producerFile.getAbsolutePath());
+			producerProcessBuilder.redirectOutput(producerFile);
+			producerProcess = producerProcessBuilder.start();
+
+			waitForLogEntryInFile("Partitioning producer", producerFile, "Started PartProducerApplication in");
+
+			ProcessBuilder consumer1Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-rabbit.jar", "--server.port=12003");
+			File consumer1File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer1File.getAbsolutePath());
+			consumer1Builder.redirectOutput(consumer1File);
+			consumer1Process = consumer1Builder.start();
+
+			ProcessBuilder consumer2Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-rabbit.jar", "--server.port=12004",
+					"--spring.cloud.stream.bindings.input.consumer.instanceIndex=1");
+			File consumer2File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer2File.getAbsolutePath());
+			consumer2Builder.redirectOutput(consumer2File);
+			consumer2Process = consumer2Builder.start();
+
+			ProcessBuilder consumer3Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-rabbit.jar", "--server.port=12005",
+					"--spring.cloud.stream.bindings.input.consumer.instanceIndex=2");
+			File consumer3File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer3File.getAbsolutePath());
+			consumer3Builder.redirectOutput(consumer3File);
+			consumer3Process = consumer3Builder.start();
+
+			ProcessBuilder consumer4Builder = new ProcessBuilder("java", "-jar", "/tmp/partitioning-consumer-rabbit.jar", "--server.port=12006",
+					"--spring.cloud.stream.bindings.input.consumer.instanceIndex=3");
+			File consumer4File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumer4File.getAbsolutePath());
+			consumer4Builder.redirectOutput(consumer4File);
+			consumer4Process = consumer4Builder.start();
+
+			Future<?> future1 = verifyPartitions("Partitioning Consumer-1", consumer1File,
+					"f received from partition partitioned.destination.myGroup-0",
+					"g received from partition partitioned.destination.myGroup-0",
+					"h received from partition partitioned.destination.myGroup-0");
+			Future<?> future2 = verifyPartitions("Partitioning Consumer-2", consumer2File,
+					"fo received from partition partitioned.destination.myGroup-1",
+					"go received from partition partitioned.destination.myGroup-1",
+					"ho received from partition partitioned.destination.myGroup-1");
+			Future<?> future3 = verifyPartitions("Partitioning Consumer-3",consumer3File,
+					"foo received from partition partitioned.destination.myGroup-2",
+					"goo received from partition partitioned.destination.myGroup-2",
+					"hoo received from partition partitioned.destination.myGroup-2");
+			Future<?> future4 = verifyPartitions("Partitioning Consumer-4",consumer4File,
+					"fooz received from partition partitioned.destination.myGroup-3",
+					"gooz received from partition partitioned.destination.myGroup-3",
+					"hooz received from partition partitioned.destination.myGroup-3");
+
+			verifyResults(future1, future2, future3, future4);
+		}
+		finally {
+			if (producerProcess != null) {
+				producerProcess.destroyForcibly();
+			}
+			if (consumer1Process != null) {
+				consumer1Process.destroyForcibly();
+			}
+			if (consumer2Process != null) {
+				consumer2Process.destroyForcibly();
+			}
+			if (consumer3Process != null) {
+				consumer3Process.destroyForcibly();
+			}
+			if (consumer4Process != null) {
+				consumer4Process.destroyForcibly();
+			}
+		}
+	}
+
+	private Future<?> verifyPartitions(String consumer1Msg, File consumer1File,
+									   String consumer2Msg, File consumer2File,
+									   String... entries) {
+
+		ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+		Future<?> submit = executorService.submit(() -> {
+			boolean found = waitForLogEntryInFileWithoutFailing(consumer1Msg, consumer1File, entries);
+			if (!found) {
+				found = waitForLogEntryInFileWithoutFailing(consumer2Msg, consumer2File, entries);
+			}
+			if (!found) {
+				fail("Could not find the test data in the logs");
+			}
+		});
+
+		executorService.shutdown();
+		return submit;
+	}
+
+	private Future<?> verifyPartitions(String consumer1Msg, File consumer1File,
+									   String... entries) {
+
+		ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+		Future<?> submit = executorService.submit(() -> {
+			boolean found = waitForLogEntryInFileWithoutFailing(consumer1Msg, consumer1File, entries);
+			if (!found) {
+				fail("Could not find the test data in the logs");
+			}
+		});
+
+		executorService.shutdown();
+		return submit;
+	}
+
+	private void verifyResults(Future<?>... futures) throws Exception {
+		for (Future<?> future : futures) {
+			try {
+				future.get();
+			}
+			catch (Exception e) {
+				throw e;
+			}
+		}
+	}
+}

spring-cloud-stream-local-acceptance-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java

@@ -0,0 +1,344 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 sample.acceptance.tests;
+
+import org.assertj.core.util.Files;
+import org.junit.After;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.SingleConnectionDataSource;
+import org.springframework.web.client.RestTemplate;
+
+import javax.sql.DataSource;
+import java.io.File;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Do not run these tests as part of an IDE build or individually.
+ * These are acceptance tests for the spring cloud stream samples.
+ * The recommended way to run these tests are using the runAcceptanceTests.sh script in this module.
+ * More about running that script can be found in the README.
+ *
+ * @author Soby Chacko
+ */
+public class SampleAcceptanceTests extends AbstractSampleTests {
+
+	private static final Logger logger = LoggerFactory.getLogger(SampleAcceptanceTests.class);
+
+	private Process process;
+
+	@After
+	public void stopTheApp() {
+		if (process != null) {
+			process.destroyForcibly();
+		}
+	}
+
+	@Test
+	public void testJdbcSourceSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/jdbc-source-kafka-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("JDBC Source", file,"Started SampleJdbcSource in");
+
+		waitForLogEntryInFile("JDBC Source", file,
+				"Data received...[{id=1, name=Bob, tag=null}, {id=2, name=Jane, tag=null}, {id=3, name=John, tag=null}]");
+	}
+
+	@Test
+	public void testJdbcSourceSampleRabbit() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/jdbc-source-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("JDBC Source", file,"Started SampleJdbcSource in");
+
+		waitForLogEntryInFile("JDBC Source", file,
+				"Data received...[{id=1, name=Bob, tag=null}, {id=2, name=Jane, tag=null}, {id=3, name=John, tag=null}]");
+	}
+
+	@Test
+	public void testJdbcSinkSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/jdbc-sink-kafka-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("JDBC Sink", file,"Started SampleJdbcSink in");
+
+		verifyJdbcSink();
+	}
+
+	@Test
+	public void testJdbcSinkSampleRabbit() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/jdbc-sink-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("JDBC Sink", file,"Started SampleJdbcSink in");
+
+		verifyJdbcSink();
+	}
+
+	@Test
+	public void testDynamicSourceSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/dynamic-destination-source-kafka-sample.jar", "--management.endpoints.web.exposure.include=*");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Dynamic Source", file,"Started SourceApplication in");
+
+		RestTemplate restTemplate = new RestTemplate();
+		restTemplate.postForObject(
+				"http://localhost:8080",
+				"{\"id\":\"customerId-1\",\"bill-pay\":\"100\"}", String.class);
+
+		waitForLogEntryInFile("Dynamic Source", file,
+				"Data received from customer-1...{\"id\":\"customerId-1\",\"bill-pay\":\"100\"}");
+
+		restTemplate.postForObject(
+				"http://localhost:8080",
+				"{\"id\":\"customerId-2\",\"bill-pay2\":\"200\"}", String.class);
+
+		waitForLogEntryInFile("Dynamic Source", file,
+				"Data received from customer-2...{\"id\":\"customerId-2\",\"bill-pay2\":\"200\"}");
+
+		Files.delete(file);
+	}
+
+	@Test
+	public void testDynamicSourceSampleRabbit() throws Exception {
+
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/dynamic-destination-source-rabbit-sample.jar", "--management.endpoints.web.exposure.include=*");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Dynamic Source", file,"Started SourceApplication in");
+
+		RestTemplate restTemplate = new RestTemplate();
+		restTemplate.postForObject(
+				"http://localhost:8080",
+				"{\"id\":\"customerId-1\",\"bill-pay\":\"100\"}", String.class);
+
+		waitForLogEntryInFile("Dynamic Source", file,
+				"Data received from customer-1...{\"id\":\"customerId-1\",\"bill-pay\":\"100\"}");
+
+		restTemplate.postForObject(
+				"http://localhost:8080",
+				"{\"id\":\"customerId-2\",\"bill-pay2\":\"200\"}", String.class);
+
+		waitForLogEntryInFile("Dynamic Source", file,
+				"Data received from customer-2...{\"id\":\"customerId-2\",\"bill-pay2\":\"200\"}");
+
+		Files.delete(file);
+	}
+
+	@Test
+	public void testMultiBinderKafkaInputRabbitOutput() throws Exception {
+
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/multibinder-kafka-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Multibinder", file,"Started MultibinderApplication in");
+
+		waitForLogEntryInFile("Multibinder", file, "Data received...bar", "Data received...foo");
+	}
+
+	@Test
+	public void testMultiBinderTwoKafkaClusters() throws Exception {
+
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/multibinder-two-kafka-clusters-sample.jar",
+				"--kafkaBroker1=localhost:9092", "--zk1=localhost:2181",
+				"--kafkaBroker2=localhost:9093", "--zk2=localhost:2182");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Multibinder 2 Kafka Clusters", file,"Started MultibinderApplication in");
+
+		waitForLogEntryInFile("Multibinder 2 Kafka Clusters", file, "Data received...bar", "Data received...foo");
+
+		Files.delete(file);
+	}
+
+	@Test
+	public void testStreamListenerBasicSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/streamlistener-basic-kafka-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Streamlistener basic", file,"Started TypeConversionApplication in");
+
+		waitForLogEntryInFile("Streamlistener basic", file,
+				"At the Source", "Sending value: {\"value\":\"hi\"}", "At the transformer",
+				"Received value hi of type class demo.Bar",
+				"Transforming the value to HI and with the type class demo.Bar",
+				"At the Sink",
+				"Received transformed message HI of type class demo.Foo");
+	}
+
+	@Test
+	public void testStreamListenerBasicSampleRabbit() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/streamlistener-basic-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Streamlistener basic", file,"Started TypeConversionApplication in");
+
+		waitForLogEntryInFile("Streamlistener basic", file,
+				"At the Source", "Sending value: {\"value\":\"hi\"}", "At the transformer",
+				"Received value hi of type class demo.Bar",
+				"Transforming the value to HI and with the type class demo.Bar",
+				"At the Sink",
+				"Received transformed message HI of type class demo.Foo");
+	}
+
+	@Test
+	public void testReactiveProcessorSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/reactive-processor-kafka-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Reactive processor", file,"Started ReactiveProcessorApplication in");
+
+		waitForLogEntryInFile("Reactive processor", file,
+				"Data received: foobarfoobarfoo",
+				"Data received: barfoobarfoobar");
+	}
+
+	@Test
+	public void testReactiveProcessorSampleRabbit() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/reactive-processor-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Reactive processor", file,"Started ReactiveProcessorApplication in");
+
+		waitForLogEntryInFile("Reactive processor", file,
+				"Data received: foobarfoobarfoo",
+				"Data received: barfoobarfoobar");
+	}
+
+	@Test
+	public void testSensorAverageReactiveSampleKafka() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/sensor-average-reactive-kafka-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Sensor average", file,"Started SensorAverageProcessorApplication in");
+
+		waitForLogEntryInFile("Sensor average", file,
+				"Data received: {\"id\":100100,\"average\":",
+				"Data received: {\"id\":100200,\"average\":", "Data received: {\"id\":100300,\"average\":");
+	}
+
+	@Test
+	public void testSensorAverageReactiveSampleRabbit() throws Exception {
+
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/sensor-average-reactive-rabbit-sample.jar");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Sensor average", file,"Started SensorAverageProcessorApplication in");
+
+		waitForLogEntryInFile("Sensor average", file,
+				"Data received: {\"id\":100100,\"average\":",
+				"Data received: {\"id\":100200,\"average\":", "Data received: {\"id\":100300,\"average\":");
+
+		Files.delete(file);
+	}
+
+	@Test
+	public void testKafkaStreamsWordCount() throws Exception {
+		ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/kafka-streams-word-count-sample.jar",
+				"--spring.cloud.stream.kafka.streams.timeWindow.length=60000");
+		File file = Files.newTemporaryFile();
+		logger.info("Output is redirected to " + file.getAbsolutePath());
+		pb.redirectOutput(file);
+		process = pb.start();
+
+		waitForLogEntryInFile("Kafka Streams WordCount", file,"Started KafkaStreamsWordCountApplication in");
+
+		waitForLogEntryInFile("Kafka Streams WordCount", file,
+				"Data received...{\"word\":\"foo\",\"count\":",
+				"Data received...{\"word\":\"bar\",\"count\":",
+				"Data received...{\"word\":\"foobar\",\"count\":",
+				"Data received...{\"word\":\"baz\",\"count\":",
+				"Data received...{\"word\":\"fox\",\"count\":");
+
+		Files.delete(file);
+	}
+
+	private void verifyJdbcSink() {
+		JdbcTemplate db;
+		DataSource dataSource = new SingleConnectionDataSource("jdbc:mariadb://localhost:3306/sample_mysql_db",
+				"root", "pwd", false);
+
+		db = new JdbcTemplate(dataSource);
+
+		long timeout = System.currentTimeMillis() + (30 * 1000);
+		boolean exists = false;
+		while (!exists && System.currentTimeMillis() < timeout) {
+			try {
+				Thread.sleep(5 * 1000);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new IllegalStateException(e.getMessage(), e);
+			}
+
+			Integer count = db.queryForObject("select count(*) from test", Integer.class);
+
+			if (count > 0) {
+				exists = true;
+			}
+		}
+		if (!exists) {
+			fail("No records found in database!");
+		}
+	}
+}

spring-cloud-stream-local-acceptance-tests/src/test/java/sample/acceptance/tests/SchemaRegistryVanillaSampleTests.java

@@ -0,0 +1,117 @@
+package sample.acceptance.tests;
+
+import org.assertj.core.util.Files;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.File;
+
+import static org.junit.Assert.fail;
+
+/**
+ * @author Soby Chacko
+ */
+public class SchemaRegistryVanillaSampleTests extends AbstractSampleTests {
+
+	private static final Logger logger = LoggerFactory.getLogger(SchemaRegistryVanillaSampleTests.class);
+
+	@Test
+	public void testSchemaRegistryVanillaKafka() throws Exception {
+		runAgainstMiddleware("/tmp/schema-registry-vanilla-registry-kafka.jar",
+				"/tmp/schema-registry-vanilla-consumer-kafka.jar",
+				"/tmp/schema-registry-vanilla-producer1-kafka.jar",
+				"/tmp/schema-registry-vanilla-producer2-kafka.jar");
+	}
+
+	@Test
+	public void testSchemaRegistryVanillaRabbit() throws Exception {
+		runAgainstMiddleware("/tmp/schema-registry-vanilla-registry-rabbit.jar",
+				"/tmp/schema-registry-vanilla-consumer-rabbit.jar",
+				"/tmp/schema-registry-vanilla-producer1-rabbit.jar",
+				"/tmp/schema-registry-vanilla-producer2-rabbit.jar");
+	}
+
+	private void runAgainstMiddleware(String registryJar, String consumerJar, String producer1Jar, String producer2Jar) throws Exception {
+
+		Process registryProcess = null;
+		Process consumerProcess = null;
+		Process producer1Process = null;
+		Process producer2Process = null;
+
+		try {
+
+			ProcessBuilder pbRegistry = new ProcessBuilder("java", "-jar", registryJar);
+			File registryFile = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + registryFile.getAbsolutePath());
+			pbRegistry.redirectOutput(registryFile);
+			registryProcess = pbRegistry.start();
+
+			waitForLogEntryInFile("Schema Registry Vanilla Server", registryFile, "Started RegistryApplication in");
+
+			ProcessBuilder pbConsumer = new ProcessBuilder("java", "-jar", consumerJar);
+			File consumerFile = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + consumerFile.getAbsolutePath());
+			pbConsumer.redirectOutput(consumerFile);
+			consumerProcess = pbConsumer.start();
+
+			waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile, "Started ConsumerApplication in");
+
+			ProcessBuilder pbProducer1 = new ProcessBuilder("java", "-jar", producer1Jar);
+			File producer1File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + producer1File.getAbsolutePath());
+			pbProducer1.redirectOutput(producer1File);
+			producer1Process = pbProducer1.start();
+
+			waitForLogEntryInFile("Schema Registry Vanilla Producer1", producer1File, "Started Producer1Application in");
+
+			ProcessBuilder pbProducer2 = new ProcessBuilder("java", "-jar", producer2Jar);
+			File producer2File = Files.newTemporaryFile();
+			logger.info("Output is redirected to " + producer2File.getAbsolutePath());
+			pbProducer2.redirectOutput(producer2File);
+			producer2Process = pbProducer2.start();
+
+			waitForLogEntryInFile("Schema Registry Vanilla Producer2", producer2File, "Started Producer2Application in");
+
+			RestTemplate restTemplate = new RestTemplate();
+
+			MultiValueMap<String, Object> parametersMap = new LinkedMultiValueMap<>();
+			parametersMap.add("id", "foobar");
+			parametersMap.add("temperature", 30);
+			parametersMap.add("acceleration", 10);
+			parametersMap.add("velocity", 20);
+
+			restTemplate.postForObject(
+					"http://localhost:9009/messagesX", parametersMap, String.class);
+
+			boolean found = waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile,
+					"{\"id\": \"foobar-v1\", \"internalTemperature\": 30.0, \"externalTemperature\": 0.0, \"acceleration\": 10.0, \"velocity\": 20.0}");
+			if (!found) {
+				fail("Could not find the test data in the logs");
+			}
+
+			restTemplate.postForObject(
+					"http://localhost:9010/messagesX", parametersMap, String.class);
+
+			waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile,
+					"{\"id\": \"foobar-v2\", \"internalTemperature\": 30.0, \"externalTemperature\": 0.0, \"acceleration\": 10.0, \"velocity\": 20.0}");
+
+		} finally {
+			if (registryProcess != null) {
+				registryProcess.destroyForcibly();
+			}
+			if (consumerProcess != null) {
+				consumerProcess.destroyForcibly();
+			}
+			if (producer1Process != null) {
+				producer1Process.destroyForcibly();
+			}
+			if (producer2Process != null) {
+				producer2Process.destroyForcibly();
+			}
+		}
+	}
+}