From ccea09f8a938c0ecae4c34b2402126b7a13fef93 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 21 Jun 2016 20:49:46 -0400 Subject: [PATCH] Remove Kafka specific test rules from core repo --- .../test/junit/kafka/EmbeddedZookeeper.java | 107 --------- .../test/junit/kafka/KafkaTestSupport.java | 203 ------------------ .../test/junit/kafka/TestKafkaCluster.java | 75 ------- 3 files changed, 385 deletions(-) delete mode 100644 spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/EmbeddedZookeeper.java delete mode 100644 spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java delete mode 100644 spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/TestKafkaCluster.java diff --git a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/EmbeddedZookeeper.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/EmbeddedZookeeper.java deleted file mode 100644 index cd72a0914..000000000 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/EmbeddedZookeeper.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - * 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.test.junit.kafka; - -import java.io.File; -import java.net.InetSocketAddress; - -import kafka.utils.TestUtils$; -import kafka.utils.Utils$; -import org.apache.zookeeper.server.NIOServerCnxnFactory; -import org.apache.zookeeper.server.ZooKeeperServer; - -/** - * 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-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java deleted file mode 100644 index 8fce7ba57..000000000 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java +++ /dev/null @@ -1,203 +0,0 @@ -/* - * 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.test.junit.kafka; - - -import java.util.Properties; - -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; -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; - - -/** - * 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-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/TestKafkaCluster.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/TestKafkaCluster.java deleted file mode 100644 index ee0c8daf9..000000000 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/TestKafkaCluster.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * 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.test.junit.kafka; - -import java.io.IOException; -import java.util.Properties; - -import kafka.server.KafkaConfig; -import kafka.server.KafkaServerStartable; -import kafka.utils.TestUtils; -import org.apache.curator.test.TestingServer; - -import org.springframework.util.SocketUtils; - -/** - * 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(); - } - -}