SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enables Kafka binding tests

Browse Source
This commit is contained in:
Marius Bogoevici
2015-07-17 14:04:22 -04:00
committed by Mark Fisher
parent 1a34f37bce
commit 0c76a1edd3
4 changed files with 51 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.travis.yml
View File

@@ -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=

7
spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java
View File

@@ -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;

2
spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/RawModeKafkaMessageBusTests.java
View File

@@ -51,7 +51,7 @@ import org.springframework.xd.test.TestUtils;
/**
* @author Marius Bogoevici
*/
@Ignore //TODO: Fix this test
public class RawModeKafkaMessageBusTests extends KafkaMessageBusTests {
@Override

77
spring-cloud-stream-bindings/spring-cloud-stream-binding-kafka/src/test/java/org/springframework/xd/dirt/integration/kafka/KafkaTestSupport.java
View File

@@ -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");
}
}