From 036f72ebd181e6fb66a509f463e2f10bbcca0743 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez?= Date: Thu, 15 Aug 2024 12:36:41 -0600 Subject: [PATCH] Support adding new listeners to Apache Kafka Fixes #8669 --- docs/modules/kafka.md | 8 ++ .../testcontainers/kafka/KafkaContainer.java | 96 +++++++++++++++---- .../kafka/CompatibleApacheKafkaImageTest.java | 26 +++++ .../kafka/KafkaContainerTest.java | 57 ++++++++--- 4 files changed, 155 insertions(+), 32 deletions(-) create mode 100644 modules/kafka/src/test/java/org/testcontainers/kafka/CompatibleApacheKafkaImageTest.java diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index cf3a5a27df6..a284163248a 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -46,6 +46,14 @@ Create a `ConfluentKafkaContainer` to use it in your tests: [Creating a ConlfuentKafkaContainer](../../modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java) inside_block:constructorWithVersion +### Using org.testcontainers.kafka.KafkaContainer + +Create a `KafkaContainer` to use it in your tests: + + +[Creating a KafkaContainer](../../modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java) inside_block:constructorWithVersion + + ## Options ### Using external Zookeeper diff --git a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java index 7eb8fa18c8b..e6f58474404 100644 --- a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java @@ -7,7 +7,10 @@ import org.testcontainers.utility.DockerImageName; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.function.Supplier; /** * Testcontainers implementation for Apache Kafka. @@ -24,11 +27,11 @@ public class KafkaContainer extends GenericContainer { private static final int KAFKA_PORT = 9092; - private static final String DEFAULT_INTERNAL_TOPIC_RF = "1"; - private static final String STARTER_SCRIPT = "/tmp/testcontainers_start.sh"; - private static final String DEFAULT_CLUSTER_ID = "4L6g3nShT-eMCtK--X86sw"; + private final Set listeners = new HashSet<>(); + + private final Set> advertisedListeners = new HashSet<>(); public KafkaContainer(String imageName) { this(DockerImageName.parse(imageName)); @@ -39,24 +42,7 @@ public KafkaContainer(DockerImageName dockerImageName) { dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME, APACHE_KAFKA_NATIVE_IMAGE_NAME); withExposedPorts(KAFKA_PORT); - withEnv("CLUSTER_ID", DEFAULT_CLUSTER_ID); - - withEnv( - "KAFKA_LISTENERS", - "PLAINTEXT://0.0.0.0:" + KAFKA_PORT + ",BROKER://0.0.0.0:9093, CONTROLLER://0.0.0.0:9094" - ); - withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"); - withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER"); - withEnv("KAFKA_PROCESS_ROLES", "broker,controller"); - withEnv("KAFKA_CONTROLLER_LISTENER_NAMES", "CONTROLLER"); - - withEnv("KAFKA_NODE_ID", "1"); - withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", DEFAULT_INTERNAL_TOPIC_RF); - withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", DEFAULT_INTERNAL_TOPIC_RF); - withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", DEFAULT_INTERNAL_TOPIC_RF); - withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", DEFAULT_INTERNAL_TOPIC_RF); - withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MAX_VALUE + ""); - withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0"); + withEnv(KafkaHelper.envVars()); withCommand("sh", "-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT); waitingFor(Wait.forLogMessage(".*Transitioning from RECOVERY to RUNNING.*", 1)); @@ -64,6 +50,8 @@ public KafkaContainer(DockerImageName dockerImageName) { @Override protected void configure() { + KafkaHelper.resolveListeners(this, this.listeners); + String firstNetworkAlias = getNetworkAliases().stream().findFirst().orElse(null); String networkAlias = getNetwork() != null ? firstNetworkAlias : "localhost"; String controllerQuorumVoters = String.format("%s@%s:9094", getEnvMap().get("KAFKA_NODE_ID"), networkAlias); @@ -80,7 +68,10 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) { List advertisedListeners = new ArrayList<>(); advertisedListeners.add("PLAINTEXT://" + getBootstrapServers()); advertisedListeners.add(brokerAdvertisedListener); + + advertisedListeners.addAll(KafkaHelper.resolveAdvertisedListeners(this.advertisedListeners)); String kafkaAdvertisedListeners = String.join(",", advertisedListeners); + String command = "#!/bin/bash\n"; // exporting KAFKA_ADVERTISED_LISTENERS with the container hostname command += String.format("export KAFKA_ADVERTISED_LISTENERS=%s\n", kafkaAdvertisedListeners); @@ -89,6 +80,69 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) { copyFileToContainer(Transferable.of(command, 0777), STARTER_SCRIPT); } + /** + * Add a listener in the format {@code host:port}. + * Host will be included as a network alias. + *

+ * Use it to register additional connections to the Kafka broker within the same container network. + *

+ * The listener will be added to the list of default listeners. + *

+ * Default listeners: + *

    + *
  • 0.0.0.0:9092
    • + *
  • 0.0.0.0:9093
    • + *
  • 0.0.0.0:9094
    • + *
+ *

+ * The listener will be added to the list of default advertised listeners. + *

+ * Default advertised listeners: + *

    + *
  • {@code container.getConfig().getHostName():9092}
    • + *
  • {@code container.getHost():container.getMappedPort(9093)}
    • + *
+ * @param listener a listener with format {@code host:port} + * @return this {@link KafkaContainer} instance + */ + public KafkaContainer withListener(String listener) { + this.listeners.add(listener); + this.advertisedListeners.add(() -> listener); + return this; + } + + /** + * Add a listener in the format {@code host:port} and a {@link Supplier} for the advertised listener. + * Host from listener will be included as a network alias. + *

+ * Use it to register additional connections to the Kafka broker from outside the container network + *

+ * The listener will be added to the list of default listeners. + *

+ * Default listeners: + *

    + *
  • 0.0.0.0:9092
    • + *
  • 0.0.0.0:9093
    • + *
  • 0.0.0.0:9094
    • + *
+ *

+ * The {@link Supplier} will be added to the list of default advertised listeners. + *

+ * Default advertised listeners: + *

    + *
  • {@code container.getConfig().getHostName():9092}
    • + *
  • {@code container.getHost():container.getMappedPort(9093)}
    • + *
+ * @param listener a supplier that will provide a listener + * @param advertisedListener a supplier that will provide a listener + * @return this {@link KafkaContainer} instance + */ + public KafkaContainer withListener(String listener, Supplier advertisedListener) { + this.listeners.add(listener); + this.advertisedListeners.add(advertisedListener); + return this; + } + public String getBootstrapServers() { return String.format("%s:%s", getHost(), getMappedPort(KAFKA_PORT)); } diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/CompatibleApacheKafkaImageTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/CompatibleApacheKafkaImageTest.java new file mode 100644 index 00000000000..b2bb31ae502 --- /dev/null +++ b/modules/kafka/src/test/java/org/testcontainers/kafka/CompatibleApacheKafkaImageTest.java @@ -0,0 +1,26 @@ +package org.testcontainers.kafka; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.testcontainers.AbstractKafka; + +@RunWith(Parameterized.class) +public class CompatibleApacheKafkaImageTest extends AbstractKafka { + + @Parameterized.Parameters(name = "{0}") + public static String[] params() { + return new String[] { "apache/kafka:3.8.0", "apache/kafka-native:3.8.0" }; + } + + @Parameterized.Parameter + public String imageName; + + @Test + public void testUsage() throws Exception { + try (KafkaContainer kafka = new KafkaContainer(this.imageName)) { + kafka.start(); + testKafkaFunctionality(kafka.getBootstrapServers()); + } + } +} diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java index 233ee1cd2d4..b1764d3b522 100644 --- a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java +++ b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java @@ -1,26 +1,61 @@ package org.testcontainers.kafka; import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; import org.testcontainers.AbstractKafka; +import org.testcontainers.KCatContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.SocatContainer; + +import static org.assertj.core.api.Assertions.assertThat; -@RunWith(Parameterized.class) public class KafkaContainerTest extends AbstractKafka { - @Parameterized.Parameters(name = "{0}") - public static String[] params() { - return new String[] { "apache/kafka:3.8.0", "apache/kafka-native:3.8.0" }; + @Test + public void testUsage() throws Exception { + try ( // constructorWithVersion { + KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0") + // } + ) { + kafka.start(); + testKafkaFunctionality(kafka.getBootstrapServers()); + } } - @Parameterized.Parameter - public String imageName; + @Test + public void testUsageWithListener() throws Exception { + try ( + Network network = Network.newNetwork(); + KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0") + .withListener("kafka:19092") + .withNetwork(network); + KCatContainer kcat = new KCatContainer().withNetwork(network) + ) { + kafka.start(); + kcat.start(); + + kcat.execInContainer("kcat", "-b", "kafka:19092", "-t", "msgs", "-P", "-l", "/data/msgs.txt"); + String stdout = kcat + .execInContainer("kcat", "-b", "kafka:19092", "-C", "-t", "msgs", "-c", "1") + .getStdout(); + + assertThat(stdout).contains("Message produced by kcat"); + } + } @Test - public void testUsage() throws Exception { - try (KafkaContainer kafka = new KafkaContainer(imageName)) { + public void testUsageWithListenerFromProxy() throws Exception { + try ( + Network network = Network.newNetwork(); + SocatContainer socat = new SocatContainer().withNetwork(network).withTarget(2000, "kafka", 19092); + KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0") + .withListener("kafka:19092", () -> socat.getHost() + ":" + socat.getMappedPort(2000)) + .withNetwork(network) + ) { + socat.start(); kafka.start(); - testKafkaFunctionality(kafka.getBootstrapServers()); + + String bootstrapServers = String.format("%s:%s", socat.getHost(), socat.getMappedPort(2000)); + testKafkaFunctionality(bootstrapServers); } } }