diff --git a/pom.xml b/pom.xml
index 7af25d3ce..471dde47d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -27,6 +27,7 @@
spring-cloud-stream-binder-kafka
spring-cloud-starter-stream-kafka
+ spring-cloud-stream-binder-kafka-test-support
diff --git a/spring-cloud-starter-stream-kafka/pom.xml b/spring-cloud-starter-stream-kafka/pom.xml
index caaf900ba..1c0b9f1f0 100644
--- a/spring-cloud-starter-stream-kafka/pom.xml
+++ b/spring-cloud-starter-stream-kafka/pom.xml
@@ -20,6 +20,7 @@
org.springframework.cloud
spring-cloud-stream-binder-kafka
+ ${project.version}
diff --git a/spring-cloud-stream-binder-kafka-test-support/pom.xml b/spring-cloud-stream-binder-kafka-test-support/pom.xml
new file mode 100644
index 000000000..1715eab82
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-test-support/pom.xml
@@ -0,0 +1,62 @@
+
+
+ 4.0.0
+
+ org.springframework.cloud
+ spring-cloud-stream-binder-kafka-parent
+ 1.1.0.BUILD-SNAPSHOT
+
+ spring-cloud-stream-binder-kafka-test-support
+ Kafka related test classes
+
+ 0.8.2.1
+ 2.6.0
+
+
+
+ junit
+ junit
+ compile
+
+
+ org.springframework.cloud
+ spring-cloud-stream-test-support-internal
+
+
+ org.springframework.boot
+ spring-boot-starter-logging
+
+
+ org.apache.curator
+ curator-recipes
+ ${curator.version}
+ true
+
+
+ org.apache.kafka
+ kafka_2.10
+ ${kafka.version}
+ true
+
+
+ org.apache.kafka
+ kafka_2.10
+ test
+ ${kafka.version}
+ true
+
+
+ org.apache.curator
+ curator-test
+ ${curator.version}
+ true
+
+
+
+ org.springframework.boot
+ spring-boot-starter-amqp
+ true
+
+
+
+
diff --git a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/EmbeddedZookeeper.java b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/EmbeddedZookeeper.java
new file mode 100644
index 000000000..edc7e2cf5
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/EmbeddedZookeeper.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2014 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.test.junit.kafka;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+
+import kafka.utils.TestUtils$;
+import kafka.utils.Utils$;
+
+/**
+ * A port of kafka.zk.EmbeddedZookeeper, compatible with Zookeeper 3.4 API
+ *
+ * @author Marius Bogoevici
+ */
+public class EmbeddedZookeeper {
+
+ private String connectString;
+
+ private File snapshotDir = TestUtils$.MODULE$.tempDir();
+
+ private File logDir = TestUtils$.MODULE$.tempDir();
+
+ private int tickTime = 500;
+
+ private final ZooKeeperServer zookeeper;
+
+ private int port;
+
+ private final NIOServerCnxnFactory factory;
+
+ public EmbeddedZookeeper(String connectString) throws Exception {
+ this.connectString = connectString;
+ port = Integer.parseInt(connectString.split(":")[1]);
+ zookeeper = new ZooKeeperServer(snapshotDir, logDir, tickTime);
+ factory = new NIOServerCnxnFactory();
+ factory.configure(new InetSocketAddress("127.0.0.1", port), 100);
+ factory.startup(zookeeper);
+ }
+
+ public String getConnectString() {
+ return connectString;
+ }
+
+ public File getSnapshotDir() {
+ return snapshotDir;
+ }
+
+ public File getLogDir() {
+ return logDir;
+ }
+
+ public int getTickTime() {
+ return tickTime;
+ }
+
+ public ZooKeeperServer getZookeeper() {
+ return zookeeper;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void shutdown() {
+ try {
+ zookeeper.shutdown();
+ }
+ catch (Exception e) {
+ // ignore exception
+ }
+ try {
+ factory.shutdown();
+ }
+ catch (Exception e) {
+ // ignore exception
+ }
+ try {
+ Utils$.MODULE$.rm(logDir);
+ }
+ catch (Exception e) {
+ // ignore exception
+ }
+ try {
+ Utils$.MODULE$.rm(snapshotDir);
+ }
+ catch (Exception e) {
+ // ignore exception
+ }
+ }
+}
+
diff --git a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java
new file mode 100644
index 000000000..0ebc238a3
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java
@@ -0,0 +1,203 @@
+/*
+ * Copyright 2014-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.
+ */
+package org.springframework.cloud.stream.binder.test.junit.kafka;
+
+import java.util.Properties;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.exception.ZkInterruptedException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Rule;
+
+import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
+import org.springframework.util.SocketUtils;
+
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.utils.SystemTime$;
+import kafka.utils.TestUtils;
+import kafka.utils.Utils;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+
+
+/**
+ * JUnit {@link Rule} that starts an embedded Kafka server (with an associated Zookeeper)
+ *
+ * @author Ilayaperumal Gopinathan
+ * @author Marius Bogoevici
+ * @since 1.1
+ */
+public class KafkaTestSupport extends AbstractExternalResourceTestSupport {
+
+ private static final Log log = LogFactory.getLog(KafkaTestSupport.class);
+
+ private static final String SCS_KAFKA_TEST_EMBEDDED = "SCS_KAFKA_TEST_EMBEDDED";
+
+ public static final boolean defaultEmbedded;
+
+ private static final String DEFAULT_ZOOKEEPER_CONNECT = "localhost:2181";
+
+ private static final String DEFAULT_KAFKA_CONNECT = "localhost:9092";
+
+ private ZkClient zkClient;
+
+ private EmbeddedZookeeper zookeeper;
+
+ private KafkaServer kafkaServer;
+
+ public final boolean embedded;
+
+ private final Properties brokerConfig = TestUtils.createBrokerConfig(0, TestUtils.choosePort(), false);
+
+ // caches previous failures to reach the external server - preventing repeated retries
+ private static boolean hasFailedAlready;
+
+ static {
+ // check if either the environment or Java property is set to use embedded tests
+ // unless the property is explicitly set to false, default to embedded
+ defaultEmbedded = !("false".equals(System.getenv(SCS_KAFKA_TEST_EMBEDDED))
+ || "false".equals(System.getProperty(SCS_KAFKA_TEST_EMBEDDED)));
+ }
+
+ public KafkaTestSupport() {
+ this(defaultEmbedded);
+ }
+
+ public KafkaTestSupport(boolean embedded) {
+ super("KAFKA");
+ this.embedded = embedded;
+ log.info(String.format("Testing with %s Kafka broker", embedded ? "embedded" : "external"));
+ }
+
+ public KafkaServer getKafkaServer() {
+ return kafkaServer;
+ }
+
+ public String getZkConnectString() {
+ if (embedded) {
+ return zookeeper.getConnectString();
+ }
+ else {
+ return DEFAULT_ZOOKEEPER_CONNECT;
+ }
+ }
+
+ public ZkClient getZkClient() {
+ return this.zkClient;
+ }
+
+ public String getBrokerAddress() {
+ if (embedded) {
+ return kafkaServer.config().hostName() + ":" + kafkaServer.config().port();
+ }
+ else {
+ return DEFAULT_KAFKA_CONNECT;
+ }
+ }
+
+ @Override
+ protected void obtainResource() throws Exception {
+ if (!hasFailedAlready) {
+ if (embedded) {
+ try {
+ log.debug("Starting Zookeeper");
+ zookeeper = new EmbeddedZookeeper("127.0.0.1:" + SocketUtils.findAvailableTcpPort());
+ log.debug("Started Zookeeper at " + zookeeper.getConnectString());
+ try {
+ int zkConnectionTimeout = 10000;
+ int zkSessionTimeout = 10000;
+ zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout,
+ ZKStringSerializer$.MODULE$);
+ }
+ catch (Exception e) {
+ zookeeper.shutdown();
+ throw e;
+ }
+ try {
+ log.debug("Creating Kafka server");
+ Properties brokerConfigProperties = brokerConfig;
+ brokerConfig.put("zookeeper.connect", zookeeper.getConnectString());
+ brokerConfig.put("auto.create.topics.enable", "false");
+ brokerConfig.put("delete.topic.enable", "true");
+ kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties),
+ SystemTime$.MODULE$);
+ log.debug("Created Kafka server at " + kafkaServer.config().hostName() + ":"
+ + kafkaServer.config().port());
+ }
+ catch (Exception e) {
+ zookeeper.shutdown();
+ zkClient.close();
+ throw e;
+ }
+ }
+ catch (Exception e) {
+ hasFailedAlready = true;
+ throw e;
+ }
+ }
+ else {
+ this.zkClient = new ZkClient(DEFAULT_ZOOKEEPER_CONNECT, 10000, 10000, ZKStringSerializer$.MODULE$);
+ if (ZkUtils.getAllBrokersInCluster(zkClient).size() == 0) {
+ hasFailedAlready = true;
+ throw new RuntimeException("Kafka server not available");
+ }
+ }
+ }
+ else {
+ throw new RuntimeException("Kafka server not available");
+ }
+ }
+
+ @Override
+ protected void cleanupResource() throws Exception {
+ if (embedded) {
+ try {
+ kafkaServer.shutdown();
+ }
+ catch (Exception e) {
+ // ignore errors on shutdown
+ log.error(e.getMessage(), e);
+ }
+ try {
+ Utils.rm(kafkaServer.config().logDirs());
+ }
+ catch (Exception e) {
+ // ignore errors on shutdown
+ log.error(e.getMessage(), e);
+ }
+ }
+ try {
+ zkClient.close();
+ }
+ catch (ZkInterruptedException e) {
+ // ignore errors on shutdown
+ log.error(e.getMessage(), e);
+ }
+ if (embedded) {
+ try {
+ zookeeper.shutdown();
+ }
+ catch (Exception e) {
+ // ignore errors on shutdown
+ log.error(e.getMessage(), e);
+ }
+ }
+ }
+
+}
+
diff --git a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/TestKafkaCluster.java b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/TestKafkaCluster.java
new file mode 100644
index 000000000..e11aa6c95
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/TestKafkaCluster.java
@@ -0,0 +1,76 @@
+/*
+ * 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.test.junit.kafka;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import org.apache.curator.test.TestingServer;
+
+import org.springframework.util.SocketUtils;
+
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServerStartable;
+import kafka.utils.TestUtils;
+
+/**
+ * A test Kafka + ZooKeeper pair for testing purposes.
+ *
+ * @author Eric Bottard
+ */
+public class TestKafkaCluster {
+
+ private KafkaServerStartable kafkaServer;
+
+ private TestingServer zkServer;
+
+ public TestKafkaCluster() {
+ try {
+ zkServer = new TestingServer(SocketUtils.findAvailableTcpPort());
+ }
+ catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ KafkaConfig config = getKafkaConfig(zkServer.getConnectString());
+ kafkaServer = new KafkaServerStartable(config);
+ kafkaServer.startup();
+ }
+
+ private static KafkaConfig getKafkaConfig(final String zkConnectString) {
+ scala.collection.Iterator propsI = TestUtils
+ .createBrokerConfigs(1, false).iterator();
+ assert propsI.hasNext();
+ Properties props = propsI.next();
+ assert props.containsKey("zookeeper.connect");
+ props.put("zookeeper.connect", zkConnectString);
+ return new KafkaConfig(props);
+ }
+
+ public String getKafkaBrokerString() {
+ return String.format("localhost:%d", kafkaServer.serverConfig().port());
+ }
+
+ public void stop() throws IOException {
+ kafkaServer.shutdown();
+ zkServer.stop();
+ }
+
+ public String getZkConnectString() {
+ return zkServer.getConnectString();
+ }
+
+}
+
diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml
index 3426dc64e..f1e9e69f9 100644
--- a/spring-cloud-stream-binder-kafka/pom.xml
+++ b/spring-cloud-stream-binder-kafka/pom.xml
@@ -46,8 +46,9 @@
org.springframework.cloud
- spring-cloud-stream-test-support-internal
+ spring-cloud-stream-binder-kafka-test-support
test
+ ${project.version}
org.springframework.integration
diff --git 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
index 08ca966f1..9f64c1bc5 100644
--- 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
@@ -16,6 +16,9 @@
package org.springframework.cloud.stream.binder.kafka;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
@@ -24,8 +27,6 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import kafka.admin.AdminUtils;
-import kafka.api.TopicMetadata;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
@@ -39,7 +40,7 @@ import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
-import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
+import org.springframework.cloud.stream.binder.test.junit.kafka.KafkaTestSupport;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -58,8 +59,8 @@ import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.fail;
+import kafka.admin.AdminUtils;
+import kafka.api.TopicMetadata;
/**
* Integration tests for the {@link KafkaMessageChannelBinder}.
diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
index b58fd7fc2..12ea4eae0 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
@@ -18,14 +18,11 @@ package org.springframework.cloud.stream.binder.kafka;
import java.util.List;
-import com.esotericsoftware.kryo.Kryo;
-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.cloud.stream.test.junit.kafka.TestKafkaCluster;
+import org.springframework.cloud.stream.binder.test.junit.kafka.TestKafkaCluster;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.KryoRegistrar;
@@ -34,6 +31,9 @@ import org.springframework.integration.kafka.support.LoggingProducerListener;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.tuple.TupleKryoRegistrar;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+
/**
* Test support class for {@link KafkaMessageChannelBinder}. Creates a binder that uses a
* test {@link TestKafkaCluster kafka cluster}.