From 0c76a1edd31b0667737697d2174ca0c62118172e Mon Sep 17 00:00:00 2001
From: Marius Bogoevici <mbogoevici@pivotal.io>
Date: Fri, 17 Jul 2015 14:04:22 -0400
Subject: [PATCH] Enables Kafka binding tests

---
 .travis.yml                                   |  1 +
 .../bus/kafka/KafkaMessageBusTests.java       |  7 +-
 .../kafka/RawModeKafkaMessageBusTests.java    |  2 +-
 .../integration/kafka/KafkaTestSupport.java   | 77 +++++++++++--------
 4 files changed, 51 insertions(+), 36 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 9acce5026..75ad73e46 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -24,5 +24,6 @@ env:
   - GIT_NAME="Dave Syer"
   - GIT_EMAIL=dsyer@pivotal.io
   - CI_DEPLOY_USERNAME=buildmaster
+  - SCS_KAFKA_TEST_EMBEDDED=true
   - secure: MG8dqc/nFtFMxOPMr6D8MSq6uNpWQ9fDeJrV5/bCQyL2dDNOK+wqQUf9YmD6e6fIWFmH99gLs0lWjx6WRPM2wkGEz+nDsBnqpRiIgZkCK2FGBLaMzNonfEIcQfvjKCI3rc4ufJCnELYo1k8tGIkDqtJtz2sk67oggkqoe+wfCRu1NWjBhgQ2By7TyUkelkH6cxRMHuX32N6Y/CNJItYb2dVM1FdlqZ3ZmneOJARenXFi1MhT/euPrcNbjB+XwvO4NhkyLDnp6Jgqk/DhTM+qWxKw/a9r7TfbIu7d2i25MhfLherUigSHPI0tBzzCvpoaPeEIv25BrWMGhqfq0ILW2W3yl/z+liomEU/aVD02Jc2Y4QMNO+4zmokPKaohOKE6btns6WizFcGq65/tlpg1M5l9SyYlfKS58YADO8NhNvXLLY4v1RGbeMrhmhsSX/j/hv+E0U0j4DDjGo1Z7GTKYtvu7u+6p7YRXiSAiukJIHEd0A9neSmC55EtVsKKlrLshxuQFWNyB4nYV4BGs+qVH+I82YhGC89k5HNFcknCCiXfdclm69n17ppG1t61FK3llGUb1EcbDVzTE2brKc2skVVUk6IOPCld1gjU9nXwo2kICR+1c2TtGLQwXwVuhB6bVMK9+56RM+6wsNhmf2HjBeMLyt8WDk5FFerGmf8kaG0=
 
diff --git a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java
index 0e9a5c270..6e382b0fe 100644
--- a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java
+++ b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java
@@ -31,8 +31,8 @@ import java.util.concurrent.TimeUnit;
 
 import kafka.api.OffsetRequest;
 
+import org.junit.ClassRule;
 import org.junit.Ignore;
-import org.junit.Rule;
 import org.junit.Test;
 
 import org.springframework.integration.channel.DirectChannel;
@@ -56,11 +56,10 @@ import org.springframework.xd.dirt.integration.kafka.KafkaTestSupport;
  * @author Eric Bottard
  * @author Marius Bogoevici
  */
-@Ignore //TODO: Fix this test
 public class KafkaMessageBusTests extends PartitionCapableBusTests {
 
-	@Rule
-	public KafkaTestSupport kafkaTestSupport = new KafkaTestSupport();
+	@ClassRule
+	public static KafkaTestSupport kafkaTestSupport = new KafkaTestSupport();
 
 	private KafkaTestMessageBus messageBus;
 
diff --git a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/RawModeKafkaMessageBusTests.java b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/RawModeKafkaMessageBusTests.java
index f0bf1f864..ebf1d4324 100644
--- a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/RawModeKafkaMessageBusTests.java
+++ b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/RawModeKafkaMessageBusTests.java
@@ -51,7 +51,7 @@ import org.springframework.xd.test.TestUtils;
 /**
  * @author Marius Bogoevici
  */
-@Ignore //TODO: Fix this test
+
 public class RawModeKafkaMessageBusTests extends KafkaMessageBusTests {
 
 	@Override
diff --git a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/kafka/KafkaTestSupport.java b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/kafka/KafkaTestSupport.java
index 88d38942a..72ca03470 100644
--- a/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/kafka/KafkaTestSupport.java
+++ b/spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/kafka/KafkaTestSupport.java
@@ -27,11 +27,12 @@ import kafka.utils.TestZKUtils;
 import kafka.utils.Utils;
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
+
 import org.I0Itec.zkclient.ZkClient;
 import org.I0Itec.zkclient.exception.ZkInterruptedException;
+import org.junit.Rule;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.junit.Rule;
 
 import org.springframework.xd.test.AbstractExternalResourceTestSupport;
 
@@ -46,7 +47,7 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	private static final Logger log = LoggerFactory.getLogger(KafkaTestSupport.class);
 
-	private static final String XD_KAFKA_TEST_EMBEDDED = "XD_KAFKA_TEST_EMBEDDED";
+	private static final String SCS_KAFKA_TEST_EMBEDDED = "SCS_KAFKA_TEST_EMBEDDED";
 
 	public static final boolean embedded;
 
@@ -62,8 +63,12 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	private Properties brokerConfig = TestUtils.createBrokerConfig(0, TestUtils.choosePort(), false);
 
+	// caches previous failures to reach the external server - preventing repeated retries
+	private static boolean hasFailedAlready = false;
+
 	static {
-		embedded = "true".equals(System.getProperty(XD_KAFKA_TEST_EMBEDDED));
+		// check if either the environment or Java property is set to use embedded tests
+		embedded = "true".equals(System.getenv(SCS_KAFKA_TEST_EMBEDDED)) || "true".equals(System.getProperty(SCS_KAFKA_TEST_EMBEDDED));
 		log.info(String.format("Testing with %s Kafka broker", embedded ? "embedded" : "external"));
 	}
 
@@ -95,36 +100,46 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	@Override
 	protected void obtainResource() throws Exception {
-		if (embedded) {
-			log.debug("Starting Zookeeper");
-			zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
-			log.debug("Started Zookeeper at " + zookeeper.getConnectString());
-			try {
-				int zkConnectionTimeout = 6000;
-				int zkSessionTimeout = 6000;
-				zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout, ZKStringSerializer$.MODULE$);
+		if (!hasFailedAlready) {
+			if (embedded) {
+				try {
+					log.debug("Starting Zookeeper");
+					zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
+					log.debug("Started Zookeeper at " + zookeeper.getConnectString());
+					try {
+						int zkConnectionTimeout = 6000;
+						int zkSessionTimeout = 6000;
+						zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout, ZKStringSerializer$.MODULE$);
+					}
+					catch (Exception e) {
+						zookeeper.shutdown();
+						throw e;
+					}
+					try {
+						log.debug("Creating Kafka server");
+						Properties brokerConfigProperties = brokerConfig;
+						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;
+				}
 			}
-			catch (Exception e) {
-				zookeeper.shutdown();
-				throw e;
-			}
-			try {
-				log.debug("Creating Kafka server");
-				Properties brokerConfigProperties = brokerConfig;
-				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;
-			}
-		}
-		else {
-			this.zkClient = new ZkClient(DEFAULT_ZOOKEEPER_CONNECT, 5000, 5000, ZKStringSerializer$.MODULE$);
-			if (ZkUtils.getAllBrokersInCluster(zkClient).size() == 0) {
-				throw new RuntimeException("Kafka server not available");
+			else {
+				this.zkClient = new ZkClient(DEFAULT_ZOOKEEPER_CONNECT, 2000, 2000, 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");
 		}
 	}