From 1f7ddd77d2da2ac512d41e766751242a8a79208a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez?= Date: Tue, 25 Jul 2023 17:21:51 -0600 Subject: [PATCH 1/9] Add KafkaContainer#withListener Allow to register additional listeners. E.g. when using along with Toxiproxy, Schema Registry, Kafka Connect, KCat. Listeners's host will register as a network aliases. --- .../containers/KafkaContainer.java | 71 ++++++++++++++++--- .../containers/KafkaContainerTest.java | 26 +++++++ 2 files changed, 87 insertions(+), 10 deletions(-) diff --git a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java index 443cdde2d87..1e686da0a19 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -6,7 +6,12 @@ import org.testcontainers.utility.ComparableVersion; import org.testcontainers.utility.DockerImageName; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; import java.util.Objects; +import java.util.Set; +import java.util.function.Supplier; /** * This container wraps Confluent Kafka and Zookeeper (optionally) @@ -36,6 +41,8 @@ public class KafkaContainer extends GenericContainer { private String clusterId = DEFAULT_CLUSTER_ID; + private static final String PROTOCOL_PREFIX = "TC"; + /** * @deprecated use {@link #KafkaContainer(DockerImageName)} instead */ @@ -56,10 +63,6 @@ public KafkaContainer(final DockerImageName dockerImageName) { super(dockerImageName); dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME); - // Use two listeners with different names, it will force Kafka to communicate with itself via internal - // listener when KAFKA_INTER_BROKER_LISTENER_NAME is set, otherwise Kafka will try to use the advertised listener - withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:" + KAFKA_PORT + ",BROKER://0.0.0.0:9092"); - withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT"); withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER"); withEnv("KAFKA_BROKER_ID", "1"); @@ -133,6 +136,37 @@ public String getBootstrapServers() { @Override protected void configure() { + // Use two listeners with different names, it will force Kafka to communicate with itself via internal + // listener when KAFKA_INTER_BROKER_LISTENER_NAME is set, otherwise Kafka will try to use the advertised listener + Set listeners = new HashSet<>(); + listeners.add("PLAINTEXT://0.0.0.0:" + KAFKA_PORT); + listeners.add("BROKER://0.0.0.0:9092"); + + Set listenerSecurityProtocolMap = new HashSet<>(); + listenerSecurityProtocolMap.add("BROKER:PLAINTEXT"); + listenerSecurityProtocolMap.add("PLAINTEXT:PLAINTEXT"); + + List> listenersToTransform = new ArrayList<>(this.listeners); + for (int i = 0; i < listenersToTransform.size(); i++) { + Supplier listenerSupplier = listenersToTransform.get(i); + String protocol = String.format("%s-%d", PROTOCOL_PREFIX, i); + String listener = listenerSupplier.get(); + String listenerPort = listener.split(":")[1]; + String listenerProtocol = String.format("%s://0.0.0.0:%s", protocol, listenerPort); + String protocolMap = String.format("%s:PLAINTEXT", protocol); + listeners.add(listenerProtocol); + listenerSecurityProtocolMap.add(protocolMap); + + String host = listener.split(":")[0]; + withNetworkAliases(host); + } + + String kafkaListeners = String.join(",", listeners); + String kafkaListenerSecurityProtocolMap = String.join(",", listenerSecurityProtocolMap); + + withEnv("KAFKA_LISTENERS", kafkaListeners); + withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", kafkaListenerSecurityProtocolMap); + if (this.kraftEnabled) { waitingFor(Wait.forLogMessage(".*Transitioning from RECOVERY to RUNNING.*", 1)); configureKraft(); @@ -180,14 +214,24 @@ protected void configureZookeeper() { protected void containerIsStarting(InspectContainerResponse containerInfo) { super.containerIsStarting(containerInfo); + List advertisedListeners = new ArrayList<>(); + advertisedListeners.add(getBootstrapServers()); + advertisedListeners.add(brokerAdvertisedListener(containerInfo)); + + List> listenersToTransform = new ArrayList<>(this.listeners); + for (int i = 0; i < listenersToTransform.size(); i++) { + Supplier listenerSupplier = listenersToTransform.get(i); + String protocol = String.format("%s-%d", PROTOCOL_PREFIX, i); + String listener = listenerSupplier.get(); + String listenerProtocol = String.format("%s://%s", protocol, listener); + advertisedListeners.add(listenerProtocol); + } + + 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,%s\n", - getBootstrapServers(), - brokerAdvertisedListener(containerInfo) - ); + command += String.format("export KAFKA_ADVERTISED_LISTENERS=%s\n", kafkaAdvertisedListeners); if (this.kraftEnabled && isLessThanCP740()) { // Optimization: skip the checks @@ -223,6 +267,13 @@ protected String commandZookeeper() { return command; } + private final Set> listeners = new HashSet<>(); + + public KafkaContainer withListener(Supplier listenerSupplier) { + this.listeners.add(listenerSupplier); + return this; + } + protected String brokerAdvertisedListener(InspectContainerResponse containerInfo) { return String.format("BROKER://%s:%s", containerInfo.getConfig().getHostName(), "9092"); } diff --git a/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java index 860c990cb32..d63573afff0 100644 --- a/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java +++ b/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java @@ -16,6 +16,7 @@ import org.junit.Test; import org.rnorth.ducttape.unreliables.Unreliables; import org.testcontainers.Testcontainers; +import org.testcontainers.images.builder.Transferable; import org.testcontainers.utility.DockerImageName; import java.time.Duration; @@ -195,6 +196,31 @@ public void testKraftPrecedenceOverEmbeddedZookeeper() throws Exception { } } + @Test + public void testUsageWithListener() throws Exception { + try ( + Network network = Network.newNetwork(); + KafkaContainer kafka = new KafkaContainer(KAFKA_KRAFT_TEST_IMAGE) + .withListener(() -> "kafka:19092") + .withNetwork(network); + GenericContainer kcat = new GenericContainer<>("confluentinc/cp-kcat:7.4.1") + .withCreateContainerCmdModifier(cmd -> { + cmd.withEntrypoint("sh"); + }) + .withCopyToContainer(Transferable.of("Message produced by kcat"), "/data/msgs.txt") + .withNetwork(network) + .withCommand("-c", "tail -f /dev/null") + ) { + 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"); + } + } + protected void testKafkaFunctionality(String bootstrapServers) throws Exception { testKafkaFunctionality(bootstrapServers, 1, 1); } From 323057914fb440da3b1d6861035d3eba1db69867 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez?= Date: Mon, 14 Aug 2023 17:13:03 -0600 Subject: [PATCH 2/9] Add docs --- docs/modules/kafka.md | 23 ++++++++++++------ .../containers/KafkaContainer.java | 24 +++++++++++++++++-- .../containers/KafkaContainerTest.java | 13 +++++----- 3 files changed, 44 insertions(+), 16 deletions(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index 13f8cf3b0a8..b4661db5913 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -42,18 +42,27 @@ KRaft mode was declared production ready in 3.3.1 (confluentinc/cp-kafka:7.3.x)" See the [versions interoperability matrix](https://docs.confluent.io/platform/current/installation/versions-interoperability.html) for more details. -## Multi-container usage +## Register listeners -If your test needs to run some other Docker container which needs access to Kafka, do the following: +There are scenarios where connecting to the broker is needed. The consumer/producer can be another container +in the same network or a different process where port to connect to is different from the exposed port `9093`. +E.g [Toxiproxy](../../docs/modules/toxiproxy.md). -* Run your other container on the same network as Kafka container, e.g.: -[Network](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:withKafkaNetwork +[Register listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:registerListener -* Use `kafka.getNetworkAliases().get(0)+":9092"` as bootstrap server location. -Or just give your Kafka container a network alias of your liking. -You will need to explicitly create a network and set it on the Kafka container as well as on your other containers that need to communicate with Kafka. +Container defined in the same network: + + +[Create kcat container](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:createKCatContainer + + +Client using the new listener registered: + + +[Produce/Consume via new listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:produceConsumeMessage + ## Adding this module to your project dependencies diff --git a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java index 09ccd324b22..8c8c1e1322f 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -50,6 +50,8 @@ public class KafkaContainer extends GenericContainer { private static final String PROTOCOL_PREFIX = "TC"; + private final Set> listeners = new HashSet<>(); + /** * @deprecated use {@link #KafkaContainer(DockerImageName)} instead */ @@ -274,8 +276,26 @@ protected String commandZookeeper() { return command; } - private final Set> listeners = new HashSet<>(); - + /** + * Add a {@link Supplier} that will provide a listener with format {@code host:port}. + * Host will be added as a network alias. + *

+ * The listener will be added to the existing listeners. + *

+ * Existing listeners: + *

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

+ * Existing advertised listeners: + *

    + *
  • {@code container.getHost():container.getMappedPort(9093)}
    • + *
  • {@code container.getConfig().getHostName():9092}
    • + *
+ * @param listenerSupplier a supplier that will provide a listener + * @return this {@link KafkaContainer} instance + */ public KafkaContainer withListener(Supplier listenerSupplier) { this.listeners.add(listenerSupplier); return this; diff --git a/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java index d63573afff0..fb7c9dacc6e 100644 --- a/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java +++ b/modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java @@ -84,16 +84,9 @@ public void testExternalZookeeperWithExternalNetwork() throws Exception { .withNetwork(network) .withNetworkAliases("zookeeper") .withEnv("ZOOKEEPER_CLIENT_PORT", "2181"); - // withKafkaNetwork { - GenericContainer application = new GenericContainer<>(DockerImageName.parse("alpine")) - .withNetwork(network) - // } - .withNetworkAliases("dummy") - .withCommand("sleep 10000") ) { zookeeper.start(); kafka.start(); - application.start(); testKafkaFunctionality(kafka.getBootstrapServers()); } @@ -200,9 +193,12 @@ public void testKraftPrecedenceOverEmbeddedZookeeper() throws Exception { public void testUsageWithListener() throws Exception { try ( Network network = Network.newNetwork(); + // registerListener { KafkaContainer kafka = new KafkaContainer(KAFKA_KRAFT_TEST_IMAGE) .withListener(() -> "kafka:19092") .withNetwork(network); + // } + // createKCatContainer { GenericContainer kcat = new GenericContainer<>("confluentinc/cp-kcat:7.4.1") .withCreateContainerCmdModifier(cmd -> { cmd.withEntrypoint("sh"); @@ -210,13 +206,16 @@ public void testUsageWithListener() throws Exception { .withCopyToContainer(Transferable.of("Message produced by kcat"), "/data/msgs.txt") .withNetwork(network) .withCommand("-c", "tail -f /dev/null") + // } ) { kafka.start(); kcat.start(); + // produceConsumeMessage { 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"); } } From ee9f7e2d604643f183c2f61093bba3591cc6a6c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:57:45 -0600 Subject: [PATCH 3/9] Update docs/modules/kafka.md Co-authored-by: Kevin Wittek --- docs/modules/kafka.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index b4661db5913..cc1b6abaccf 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -45,7 +45,7 @@ See the [versions interoperability matrix](https://docs.confluent.io/platform/cu ## Register listeners There are scenarios where connecting to the broker is needed. The consumer/producer can be another container -in the same network or a different process where port to connect to is different from the exposed port `9093`. +in the same network or a different process where the port to connect differs from the default exposed port `9093`. E.g [Toxiproxy](../../docs/modules/toxiproxy.md). From 1cd41b2facf8884c877135354b9bbf3df033ba18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:58:09 -0600 Subject: [PATCH 4/9] Update docs/modules/kafka.md Co-authored-by: Kevin Wittek --- docs/modules/kafka.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index cc1b6abaccf..deddf406820 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -58,7 +58,7 @@ Container defined in the same network: [Create kcat container](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:createKCatContainer -Client using the new listener registered: +Client using the new registered listener: [Produce/Consume via new listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:produceConsumeMessage From 582ff2d79d17b2b861e8e1d8a5aa10afc75beca0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:58:14 -0600 Subject: [PATCH 5/9] Update modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java Co-authored-by: Kevin Wittek --- .../main/java/org/testcontainers/containers/KafkaContainer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java index 8c8c1e1322f..805fef8b0dd 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -280,7 +280,7 @@ protected String commandZookeeper() { * Add a {@link Supplier} that will provide a listener with format {@code host:port}. * Host will be added as a network alias. *

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

* Existing listeners: *

    From d562a9ae86404d47d4b05bcd0dca650c0ecd1cbf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:58:24 -0600 Subject: [PATCH 6/9] Update modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java Co-authored-by: Kevin Wittek --- .../main/java/org/testcontainers/containers/KafkaContainer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java index 805fef8b0dd..f03149e252d 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -288,7 +288,7 @@ protected String commandZookeeper() { *
  • 0.0.0.0:9093
    • *
*

- * Existing advertised listeners: + * Default advertised listeners: *

    *
  • {@code container.getHost():container.getMappedPort(9093)}
    • *
  • {@code container.getConfig().getHostName():9092}
    • From a1748f011c70d66fa7f3e2ca4c18fc03ddc295a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:58:40 -0600 Subject: [PATCH 7/9] Update docs/modules/kafka.md Co-authored-by: Kevin Wittek --- docs/modules/kafka.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index deddf406820..274054cf5ca 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -49,7 +49,7 @@ in the same network or a different process where the port to connect differs fro E.g [Toxiproxy](../../docs/modules/toxiproxy.md). -[Register listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:registerListener +[Register additional listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:registerListener Container defined in the same network: From 3b94b493242497b39e3d2f1fa8736057e45f613e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?= Date: Wed, 16 Aug 2023 15:58:47 -0600 Subject: [PATCH 8/9] Update modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java Co-authored-by: Kevin Wittek --- .../main/java/org/testcontainers/containers/KafkaContainer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java index f03149e252d..00d8e2998cc 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -282,7 +282,7 @@ protected String commandZookeeper() { *

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

    - * Existing listeners: + * Default listeners: *

      *
    • 0.0.0.0:9092
      • *
    • 0.0.0.0:9093
      • From 12682a47e1b0020158dd9a349f970ea4ab4d2786 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez?= Date: Wed, 16 Aug 2023 17:55:31 -0600 Subject: [PATCH 9/9] Polish docs --- docs/modules/kafka.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index 274054cf5ca..41e6f2c03e9 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -44,9 +44,9 @@ See the [versions interoperability matrix](https://docs.confluent.io/platform/cu ## Register listeners -There are scenarios where connecting to the broker is needed. The consumer/producer can be another container -in the same network or a different process where the port to connect differs from the default exposed port `9093`. -E.g [Toxiproxy](../../docs/modules/toxiproxy.md). +There are scenarios where additional listeners are needed because the consumer/producer can be in another +container in the same network or a different process where the port to connect differs from the default +exposed port `9093`. E.g [Toxiproxy](../../docs/modules/toxiproxy.md). [Register additional listener](../../modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java) inside_block:registerListener