From 910233ec4cc812dfe9e851f53d764788285602b7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 30 Sep 2019 13:20:05 -0400 Subject: [PATCH] GH-1254: Configurable Embedded Zookeeper Port Resolves https://github.com/spring-projects/spring-kafka/issues/1254 - port `EmbeddedKafkaBroker` from scala and add ZK port configuration - add `zookeeperPort` to `@EmbeddedKafka` --- .../kafka/test/EmbeddedKafkaBroker.java | 130 +++++++++++++++++- .../condition/EmbeddedKafkaCondition.java | 6 +- .../kafka/test/context/EmbeddedKafka.java | 7 + .../EmbeddedKafkaContextCustomizer.java | 10 +- .../kafka/test/rule/EmbeddedKafkaRule.java | 5 + .../kafka/test/EmbeddedKafkaBrokerTests.java | 39 ++++++ .../rule/AddressableEmbeddedBrokerTests.java | 17 ++- 7 files changed, 198 insertions(+), 16 deletions(-) create mode 100644 spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index 08b99416..14e9d13b 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java @@ -18,6 +18,9 @@ package org.springframework.kafka.test; import static org.assertj.core.api.Assertions.assertThat; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; @@ -46,6 +49,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.security.auth.SecurityProtocol; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; +import org.apache.zookeeper.server.NIOServerCnxnFactory; +import org.apache.zookeeper.server.ZooKeeperServer; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; @@ -63,7 +69,7 @@ import kafka.server.NotRunning; import kafka.utils.CoreUtils; import kafka.utils.TestUtils; import kafka.utils.ZKStringSerializer$; -import kafka.zk.EmbeddedZookeeper; +import kafka.zk.ZkFourLetterWords; /** * An embedded Kafka Broker(s) and Zookeeper manager. @@ -114,6 +120,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { private String zkConnect; + private int zkPort; + private int[] kafkaPorts; private Duration adminTimeout = DEFAULT_ADMIN_TIMEOUT; @@ -190,6 +198,16 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { return this; } + /** + * Set an explicit port for the embedded Zookeeper. + * @param port the port. + * @return the {@link EmbeddedKafkaBroker}. + * @since 2.3 + */ + public EmbeddedKafkaBroker zkPort(int port) { + this.zkPort = port; + return this; + } /** * Set the timeout in seconds for admin operations (e.g. topic creation, close). * Default 30 seconds. @@ -211,13 +229,36 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { return this; } + /** + * Get the port that the embedded Zookeeper is running on or will run on. + * @param zkPort the port. + * @since 2.3 + */ + public int getZkPort() { + return this.zookeeper != null ? this.zookeeper.getPort() : this.zkPort; + } + + /** + * Set the port to run the embedded Zookeeper on (default random). + * @param zkPort the port. + * @since 2.3 + */ + public void setZkPort(int zkPort) { + this.zkPort = zkPort; + } + @Override public void afterPropertiesSet() { - this.zookeeper = new EmbeddedZookeeper(); + try { + this.zookeeper = new EmbeddedZookeeper(this.zkPort); + } + catch (IOException | InterruptedException e) { + throw new IllegalStateException("Failed to create embedded Zookeeper", e); + } int zkConnectionTimeout = 6000; // NOSONAR magic # int zkSessionTimeout = 6000; // NOSONAR magic # - this.zkConnect = "127.0.0.1:" + this.zookeeper.port(); + this.zkConnect = "127.0.0.1:" + this.zookeeper.getPort(); this.zookeeperClient = new ZkClient(this.zkConnect, zkSessionTimeout, zkConnectionTimeout, ZKStringSerializer$.MODULE$); this.kafkaServers.clear(); @@ -360,6 +401,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { } try { this.zookeeper.shutdown(); + this.zkConnect = null; } catch (Exception e) { // do nothing @@ -510,4 +552,86 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { logger.debug("Subscription Initiated"); } + /** + * Ported from scala to allow setting the port. + * + * @author Gary Russell + * @since 2.3 + */ + public static final class EmbeddedZookeeper { + + private final NIOServerCnxnFactory factory; + + private final ZooKeeperServer zookeeper; + + private final int port; + + private final File snapshotDir; + + private final File logDir; + + public EmbeddedZookeeper(int zkPort) throws IOException, InterruptedException { + this.snapshotDir = TestUtils.tempDir(); + this.logDir = TestUtils.tempDir(); + int tickTime = 800; // allow a maxSessionTimeout of 20 * 800ms = 16 secs + + System.setProperty("zookeeper.forceSync", "no"); // disable fsync to ZK txn + // log in tests to avoid + // timeout + this.zookeeper = new ZooKeeperServer(this.snapshotDir, this.logDir, tickTime); + this.factory = new NIOServerCnxnFactory(); + InetSocketAddress addr = new InetSocketAddress("127.0.0.1", zkPort == 0 ? TestUtils.RandomPort() : zkPort); + this.factory.configure(addr, 0); + this.factory.startup(zookeeper); + this.port = zookeeper.getClientPort(); + } + + public int getPort() { + return this.port; + } + + public File getSnapshotDir() { + return this.snapshotDir; + } + + public File getLogDir() { + return this.logDir; + } + + public void shutdown() throws IOException { + // Also shuts down ZooKeeperServer + try { + this.factory.shutdown(); + } + catch (Exception e) { + logger.error(e, "ZK shutdown failed"); + } + + int n = 0; + while (n++ < 100) { + try { + ZkFourLetterWords.sendStat("127.0.0.1", port, 3000); + Thread.sleep(100); + } + catch (@SuppressWarnings("unused") Exception e) { + break; + } + } + if (n == 100) { + logger.debug("Zookeeper failed to stop"); + } + + try { + this.zookeeper.getZKDatabase().close(); + } + catch (Exception e) { + logger.error(e, "ZK db close failed"); + } + + Utils.delete(this.logDir); + Utils.delete(this.snapshotDir); + } + + } + } diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java index 2b4fd18b..79478083 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java @@ -115,9 +115,9 @@ public class EmbeddedKafkaCondition implements ExecutionCondition, AfterAllCallb @SuppressWarnings("unchecked") private EmbeddedKafkaBroker createBroker(EmbeddedKafka embedded) { EmbeddedKafkaBroker broker; - broker = new EmbeddedKafkaBroker(embedded.count(), - embedded.controlledShutdown(), embedded.topics()); - broker.kafkaPorts(embedded.ports()); + broker = new EmbeddedKafkaBroker(embedded.count(), embedded.controlledShutdown(), embedded.topics()) + .zkPort(embedded.zookeeperPort()) + .kafkaPorts(embedded.ports()); Properties properties = new Properties(); for (String pair : embedded.brokerProperties()) { diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java index 8aefb059..8d5830eb 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java @@ -95,6 +95,13 @@ public @interface EmbeddedKafka { */ int[] ports() default {0}; + /** + * Set the port on which the embedded Zookeeper should listen; + * @return the port. + * @since 2.3 + */ + int zookeeperPort() default 0; + /** * @return partitions per topic */ diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java index 9f8ab005..93ba3e11 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java @@ -67,11 +67,11 @@ class EmbeddedKafkaContextCustomizer implements ContextCustomizer { .toArray(String[]::new); EmbeddedKafkaBroker embeddedKafkaBroker = new EmbeddedKafkaBroker(this.embeddedKafka.count(), - this.embeddedKafka.controlledShutdown(), - this.embeddedKafka.partitions(), - topics); - - embeddedKafkaBroker.kafkaPorts(this.embeddedKafka.ports()); + this.embeddedKafka.controlledShutdown(), + this.embeddedKafka.partitions(), + topics) + .kafkaPorts(this.embeddedKafka.ports()) + .zkPort(this.embeddedKafka.zookeeperPort()); Properties properties = new Properties(); diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/EmbeddedKafkaRule.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/EmbeddedKafkaRule.java index 75493e72..5698c7ac 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/EmbeddedKafkaRule.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/EmbeddedKafkaRule.java @@ -96,6 +96,11 @@ public class EmbeddedKafkaRule extends ExternalResource implements TestRule { return this; } + public EmbeddedKafkaRule zkPort(int port) { + this.embeddedKafka.setZkPort(port); + return this; + } + /** * Return an underlying delegator {@link EmbeddedKafkaBroker} instance. * @return the {@link EmbeddedKafkaBroker} instance. diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java new file mode 100644 index 00000000..2d97bf26 --- /dev/null +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java @@ -0,0 +1,39 @@ +/* + * Copyright 2019 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 + * + * https://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.kafka.test; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +/** + * @author Gary Russell + * @since 2.3 + * + */ +public class EmbeddedKafkaBrokerTests { + + @Test + void testUpDown() { + EmbeddedKafkaBroker kafka = new EmbeddedKafkaBroker(1); + kafka.afterPropertiesSet(); + assertThat(kafka.getZookeeperConnectionString()).startsWith("127"); + kafka.destroy(); + assertThat(kafka.getZookeeperConnectionString()).isNull(); + } + +} diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java index 848ec75c..7bcda628 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java @@ -61,7 +61,8 @@ public class AddressableEmbeddedBrokerTests { @Test public void testKafkaEmbedded() { - assertThat(broker.getBrokersAsString()).isEqualTo("127.0.0.1:" + this.config.port); + assertThat(broker.getBrokersAsString()).isEqualTo("127.0.0.1:" + this.config.kafkaPort); + assertThat(broker.getZkPort()).isEqualTo(this.config.zkPort); assertThat(broker.getBrokersAsString()) .isEqualTo(System.getProperty(EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS)); assertThat(broker.getZookeeperConnectionString()) @@ -69,7 +70,7 @@ public class AddressableEmbeddedBrokerTests { } @Test - public void testLateStartedConsumer() throws Exception { + public void testLateStartedConsumer() { Map consumerProps = KafkaTestUtils.consumerProps(TEST_EMBEDDED, "false", this.broker); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); Consumer consumer = new KafkaConsumer<>(consumerProps); @@ -93,16 +94,22 @@ public class AddressableEmbeddedBrokerTests { @Configuration public static class Config { - private int port; + private int kafkaPort; + + private int zkPort; @Bean public EmbeddedKafkaBroker broker() throws IOException { ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(0); - this.port = ss.getLocalPort(); + this.kafkaPort = ss.getLocalPort(); + ss.close(); + ss = ServerSocketFactory.getDefault().createServerSocket(0); + this.zkPort = ss.getLocalPort(); ss.close(); return new EmbeddedKafkaBroker(1, true, TEST_EMBEDDED) - .kafkaPorts(this.port); + .zkPort(this.zkPort) + .kafkaPorts(this.kafkaPort); } }