+ * 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.apache.kafka.streams.integration.utils;
+
+import kafka.server.KafkaConfig$;
+import kafka.zk.EmbeddedZookeeper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Properties;
+import org.junit.rules.ExternalResource;
+
+/**
+ * Runs an in-memory, "embedded" Kafka cluster with 1 ZooKeeper instance and 1 Kafka broker.
+ */
+public class EmbeddedSingleNodeKafkaCluster extends ExternalResource {
+
+ private static final Logger log = LoggerFactory.getLogger(EmbeddedSingleNodeKafkaCluster.class);
+ private static final int DEFAULT_BROKER_PORT = 0; // 0 results in a random port being selected
+ private EmbeddedZookeeper zookeeper = null;
+ private KafkaEmbedded broker = null;
+
+ /**
+ * Creates and starts a Kafka cluster.
+ */
+ public void start() throws IOException, InterruptedException {
+ Properties brokerConfig = new Properties();
+
+ log.debug("Initiating embedded Kafka cluster startup");
+ log.debug("Starting a ZooKeeper instance");
+ zookeeper = new EmbeddedZookeeper();
+ log.debug("ZooKeeper instance is running at {}", zKConnectString());
+ brokerConfig.put(KafkaConfig$.MODULE$.ZkConnectProp(), zKConnectString());
+ brokerConfig.put(KafkaConfig$.MODULE$.PortProp(), DEFAULT_BROKER_PORT);
+
+ log.debug("Starting a Kafka instance on port {} ...", brokerConfig.getProperty(KafkaConfig$.MODULE$.PortProp()));
+ broker = new KafkaEmbedded(brokerConfig);
+
+ log.debug("Kafka instance is running at {}, connected to ZooKeeper at {}",
+ broker.brokerList(), broker.zookeeperConnect());
+ }
+
+ /**
+ * Stop the Kafka cluster.
+ */
+ public void stop() {
+ broker.stop();
+ zookeeper.shutdown();
+ }
+
+ /**
+ * The ZooKeeper connection string aka `zookeeper.connect` in `hostnameOrIp:port` format.
+ * Example: `127.0.0.1:2181`.
+ *
+ * You can use this to e.g. tell Kafka brokers how to connect to this instance.
+ */
+ public String zKConnectString() {
+ return "localhost:" + zookeeper.port();
+ }
+
+ /**
+ * This cluster's `bootstrap.servers` value. Example: `127.0.0.1:9092`.
+ *
+ * You can use this to tell Kafka producers how to connect to this cluster.
+ */
+ public String bootstrapServers() {
+ return broker.brokerList();
+ }
+
+ protected void before() throws Throwable {
+ start();
+ }
+
+ protected void after() {
+ stop();
+ }
+
+ /**
+ * Create a Kafka topic with 1 partition and a replication factor of 1.
+ *
+ * @param topic The name of the topic.
+ */
+ public void createTopic(String topic) {
+ createTopic(topic, 1, 1, new Properties());
+ }
+
+ /**
+ * Create a Kafka topic with the given parameters.
+ *
+ * @param topic The name of the topic.
+ * @param partitions The number of partitions for this topic.
+ * @param replication The replication factor for (the partitions of) this topic.
+ */
+ public void createTopic(String topic, int partitions, int replication) {
+ createTopic(topic, partitions, replication, new Properties());
+ }
+
+ /**
+ * Create a Kafka topic with the given parameters.
+ *
+ * @param topic The name of the topic.
+ * @param partitions The number of partitions for this topic.
+ * @param replication The replication factor for (partitions of) this topic.
+ * @param topicConfig Additional topic-level configuration settings.
+ */
+ public void createTopic(String topic,
+ int partitions,
+ int replication,
+ Properties topicConfig) {
+ broker.createTopic(topic, partitions, replication, topicConfig);
+ }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
new file mode 100644
index 0000000000000..89fe0c4ef943e
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
@@ -0,0 +1,157 @@
+/**
+ * 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.
+ */
+
+package org.apache.kafka.streams.integration.utils;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Utility functions to make integration testing more convenient.
+ */
+public class IntegrationTestUtils {
+
+ private static final int UNLIMITED_MESSAGES = -1;
+
+ /**
+ * Returns up to `maxMessages` message-values from the topic.
+ *
+ * @param topic Kafka topic to read messages from
+ * @param consumerConfig Kafka consumer configuration
+ * @param maxMessages Maximum number of messages to read via the consumer.
+ * @return The values retrieved via the consumer.
+ */
+ public static
+ * 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.apache.kafka.streams.integration.utils;
+
+
+import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.ZkConnection;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.List;
+
+import kafka.admin.AdminUtils;
+import kafka.admin.RackAwareMode;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.KafkaServer;
+import kafka.utils.CoreUtils;
+import kafka.utils.SystemTime$;
+import kafka.utils.TestUtils;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+import org.junit.rules.TemporaryFolder;
+/**
+ * Runs an in-memory, "embedded" instance of a Kafka broker, which listens at `127.0.0.1:9092` by
+ * default.
+ *
+ * Requires a running ZooKeeper instance to connect to.
+ */
+public class KafkaEmbedded {
+
+ private static final Logger log = LoggerFactory.getLogger(KafkaEmbedded.class);
+
+ private static final String DEFAULT_ZK_CONNECT = "127.0.0.1:2181";
+ private static final int DEFAULT_ZK_SESSION_TIMEOUT_MS = 10 * 1000;
+ private static final int DEFAULT_ZK_CONNECTION_TIMEOUT_MS = 8 * 1000;
+ private final Properties effectiveConfig;
+ private final File logDir;
+ public final TemporaryFolder tmpFolder;
+ private final KafkaServer kafka;
+
+ /**
+ * Creates and starts an embedded Kafka broker.
+ * @param config Broker configuration settings. Used to modify, for example, on which port the
+ * broker should listen to. Note that you cannot change the `log.dirs` setting
+ * currently.
+ */
+ public KafkaEmbedded(Properties config) throws IOException {
+ tmpFolder = new TemporaryFolder();
+ tmpFolder.create();
+ logDir = tmpFolder.newFolder();
+ effectiveConfig = effectiveConfigFrom(config);
+ boolean loggingEnabled = true;
+ KafkaConfig kafkaConfig = new KafkaConfig(effectiveConfig, loggingEnabled);
+ log.debug("Starting embedded Kafka broker (with log.dirs={} and ZK ensemble at {}) ...",
+ logDir, zookeeperConnect());
+ kafka = TestUtils.createServer(kafkaConfig, SystemTime$.MODULE$);
+ log.debug("Startup of embedded Kafka broker at {} completed (with ZK ensemble at {}) ...",
+ brokerList(), zookeeperConnect());
+ }
+
+
+ /**
+ * Creates the configuration for starting the Kafka broker by merging default values with
+ * overwrites.
+ * @param initialConfig Broker configuration settings that override the default config.
+ * @return
+ * @throws IOException
+ */
+ private Properties effectiveConfigFrom(Properties initialConfig) throws IOException {
+ Properties effectiveConfig = new Properties();
+ effectiveConfig.put(KafkaConfig$.MODULE$.BrokerIdProp(), 0);
+ effectiveConfig.put(KafkaConfig$.MODULE$.HostNameProp(), "127.0.0.1");
+ effectiveConfig.put(KafkaConfig$.MODULE$.PortProp(), "9092");
+ effectiveConfig.put(KafkaConfig$.MODULE$.NumPartitionsProp(), 1);
+ effectiveConfig.put(KafkaConfig$.MODULE$.AutoCreateTopicsEnableProp(), true);
+ effectiveConfig.put(KafkaConfig$.MODULE$.MessageMaxBytesProp(), 1000000);
+ effectiveConfig.put(KafkaConfig$.MODULE$.ControlledShutdownEnableProp(), true);
+
+ effectiveConfig.putAll(initialConfig);
+ effectiveConfig.setProperty(KafkaConfig$.MODULE$.LogDirProp(), logDir.getAbsolutePath());
+ return effectiveConfig;
+ }
+
+ /**
+ * This broker's `metadata.broker.list` value. Example: `127.0.0.1:9092`.
+ *
+ * You can use this to tell Kafka producers and consumers how to connect to this instance.
+ */
+ public String brokerList() {
+ return kafka.config().hostName() + ":" + kafka.boundPort(SecurityProtocol.PLAINTEXT);
+ }
+
+
+ /**
+ * The ZooKeeper connection string aka `zookeeper.connect`.
+ */
+ public String zookeeperConnect() {
+ return effectiveConfig.getProperty("zookeeper.connect", DEFAULT_ZK_CONNECT);
+ }
+
+ /**
+ * Stop the broker.
+ */
+ public void stop() {
+ log.debug("Shutting down embedded Kafka broker at {} (with ZK ensemble at {}) ...",
+ brokerList(), zookeeperConnect());
+ kafka.shutdown();
+ kafka.awaitShutdown();
+ log.debug("Removing logs.dir at {} ...", logDir);
+ List