diff --git a/spring-cloud-stream-binder-kafka-test-support/pom.xml b/spring-cloud-stream-binder-kafka-test-support/pom.xml index 1715eab82..452f44f09 100644 --- a/spring-cloud-stream-binder-kafka-test-support/pom.xml +++ b/spring-cloud-stream-binder-kafka-test-support/pom.xml @@ -8,10 +8,7 @@ spring-cloud-stream-binder-kafka-test-support Kafka related test classes - - 0.8.2.1 - 2.6.0 - + junit @@ -19,44 +16,9 @@ compile - org.springframework.cloud - spring-cloud-stream-test-support-internal + org.springframework.kafka + spring-kafka-test + ${spring-kafka.version} - - 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 deleted file mode 100644 index a3921b3eb..000000000 --- a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/EmbeddedZookeeper.java +++ /dev/null @@ -1,108 +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.binder.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-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 deleted file mode 100644 index 9184f2467..000000000 --- a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/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.binder.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 this.kafkaServer; - } - - public String getZkConnectString() { - if (this.embedded) { - return this.zookeeper.getConnectString(); - } - else { - return DEFAULT_ZOOKEEPER_CONNECT; - } - } - - public ZkClient getZkClient() { - return this.zkClient; - } - - public String getBrokerAddress() { - if (this.embedded) { - return this.kafkaServer.config().hostName() + ":" + this.kafkaServer.config().port(); - } - else { - return DEFAULT_KAFKA_CONNECT; - } - } - - @Override - protected void obtainResource() throws Exception { - if (!hasFailedAlready) { - if (this.embedded) { - try { - log.debug("Starting Zookeeper"); - this.zookeeper = new EmbeddedZookeeper("127.0.0.1:" + SocketUtils.findAvailableTcpPort()); - log.debug("Started Zookeeper at " + this.zookeeper.getConnectString()); - try { - int zkConnectionTimeout = 10000; - int zkSessionTimeout = 10000; - this.zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout, - ZKStringSerializer$.MODULE$); - } - catch (Exception e) { - this.zookeeper.shutdown(); - throw e; - } - try { - log.debug("Creating Kafka server"); - Properties brokerConfigProperties = this.brokerConfig; - this.brokerConfig.put("zookeeper.connect", this.zookeeper.getConnectString()); - this.brokerConfig.put("auto.create.topics.enable", "false"); - this.brokerConfig.put("delete.topic.enable", "true"); - this.kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), - SystemTime$.MODULE$); - log.debug("Created Kafka server at " + this.kafkaServer.config().hostName() + ":" - + this.kafkaServer.config().port()); - } - catch (Exception e) { - this.zookeeper.shutdown(); - this.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(this.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 (this.embedded) { - try { - this.kafkaServer.shutdown(); - } - catch (Exception e) { - // ignore errors on shutdown - log.error(e.getMessage(), e); - } - try { - Utils.rm(this.kafkaServer.config().logDirs()); - } - catch (Exception e) { - // ignore errors on shutdown - log.error(e.getMessage(), e); - } - } - try { - this.zkClient.close(); - } - catch (ZkInterruptedException e) { - // ignore errors on shutdown - log.error(e.getMessage(), e); - } - if (this.embedded) { - try { - this.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 deleted file mode 100644 index e088e6cdf..000000000 --- a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/TestKafkaCluster.java +++ /dev/null @@ -1,76 +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.binder.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(); - } - -} - 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 59070df18..e551bd8b0 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 @@ -25,7 +25,6 @@ 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.binder.test.junit.kafka.TestKafkaCluster; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.codec.Codec; import org.springframework.integration.codec.kryo.KryoRegistrar; @@ -35,8 +34,7 @@ import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; /** - * Test support class for {@link KafkaMessageChannelBinder}. Creates a binder that uses a - * test {@link TestKafkaCluster kafka cluster}. + * Test support class for {@link KafkaMessageChannelBinder}. * @author Eric Bottard * @author Marius Bogoevici * @author David Turanski