From 487212c151cef94c185d9d8b6d8ba5c2a8b44309 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 21 Jun 2016 20:48:56 -0400 Subject: [PATCH] Migrating kafka specific test classes Moving Kafka test rules from SCS over here polishing --- pom.xml | 1 + spring-cloud-starter-stream-kafka/pom.xml | 1 + .../pom.xml | 62 ++++++ .../test/junit/kafka/EmbeddedZookeeper.java | 108 ++++++++++ .../test/junit/kafka/KafkaTestSupport.java | 203 ++++++++++++++++++ .../test/junit/kafka/TestKafkaCluster.java | 76 +++++++ spring-cloud-stream-binder-kafka/pom.xml | 3 +- .../stream/binder/kafka/KafkaBinderTests.java | 11 +- .../stream/binder/kafka/KafkaTestBinder.java | 8 +- 9 files changed, 463 insertions(+), 10 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-test-support/pom.xml create mode 100644 spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/EmbeddedZookeeper.java create mode 100644 spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java create mode 100644 spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/TestKafkaCluster.java 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}.