kafka:9092 using withNetworkAliases

docs/modules/kafka.md

@@ -54,6 +54,18 @@ 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
 <!--/codeinclude-->
 
+### withNetworkAliases
+
+The first alias host name defined using `withNetworkAliases` configures port 9092 on the network set by `withNetwork`. Example for `kafka:9092`:
+
+<!--codeinclude-->
+[Creating a KafkaContainer](../../modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java) inside_block:registerAlias
+<!--/codeinclude-->
+
+This works with `org.testcontainers.kafka.KafkaContainer` and `org.testcontainers.kafka.ConfluentKafkaContainer` but not with deprecated `org.testcontainers.containers.KafkaContainer`.
+
+It only works if `withListener` is not used and `KAFKA_BROKER_ID` environment variable is not set.
+
 ## Options
         
 ### <a name="zookeeper"></a> Using external Zookeeper

modules/kafka/src/main/java/org/testcontainers/kafka/ConfluentKafkaContainer.java

@@ -48,13 +48,22 @@ protected void configure() {
 
     @Override
     protected void containerIsStarting(InspectContainerResponse containerInfo) {
+        List<String> networkAliases = getNetworkAliases();
+        String plaintextAdvertisedListener;
+        if (!getEnvMap().containsKey("KAFKA_BROKER_ID") && listeners.isEmpty() && networkAliases.size() > 1) {
+            // 0 is the random network alias generated by GenericContainer
+            plaintextAdvertisedListener = "PLAINTEXT://" + networkAliases.get(1) + ":" + KafkaHelper.KAFKA_PORT;
+        } else {
+            plaintextAdvertisedListener = "PLAINTEXT://" + getBootstrapServers();
+        }
+
         String brokerAdvertisedListener = String.format(
             "BROKER://%s:%s",
             containerInfo.getConfig().getHostName(),
             "9093"
         );
         List<String> advertisedListeners = new ArrayList<>();
-        advertisedListeners.add("PLAINTEXT://" + getBootstrapServers());
+        advertisedListeners.add(plaintextAdvertisedListener);
         advertisedListeners.add(brokerAdvertisedListener);
 
         advertisedListeners.addAll(KafkaHelper.resolveAdvertisedListeners(this.advertisedListeners));
@@ -87,8 +96,15 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) {
      * <p>
      * Default advertised listeners:
      * <ul>
-     *      <li>{@code container.getConfig().getHostName():9092}</li>
-     *      <li>{@code container.getHost():container.getMappedPort(9093)}</li>
+     *      <li>{@code container.getConfig().getHostName():container.getMappedPort(9092)}</li>
+     *      <li>{@code container.getHost():9093}</li>
+     * </ul>
+     * <p>
+     * Default advertised listeners if {@code withNetworkAliases} is used, {@code withListener} is not used,
+     * and {@code KAFKA_BROKER_ID} environment variable is not set:
+     * <ul>
+     *      <li>{@code first network alias:9092}</li>
+     *      <li>{@code container.getHost():9093}</li>
      * </ul>
      * @param listener a listener with format {@code host:port}
      * @return this {@link ConfluentKafkaContainer} instance
@@ -118,8 +134,15 @@ public ConfluentKafkaContainer withListener(String listener) {
      * <p>
      * Default advertised listeners:
      * <ul>
-     *      <li>{@code container.getConfig().getHostName():9092}</li>
-     *      <li>{@code container.getHost():container.getMappedPort(9093)}</li>
+     *      <li>{@code container.getConfig().getHostName():container.getMappedPort(9092)}</li>
+     *      <li>{@code container.getHost():9093}</li>
+     * </ul>
+     * <p>
+     * Default advertised listeners if {@code withNetworkAliases} is used, {@code withListener} is not used,
+     * and {@code KAFKA_BROKER_ID} environment variable is not set:
+     * <ul>
+     *      <li>{@code first network alias:9092}</li>
+     *      <li>{@code container.getHost():9093}</li>
      * </ul>
      * @param listener a supplier that will provide a listener
      * @param advertisedListener a supplier that will provide a listener

modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java

@@ -54,18 +54,27 @@ protected void configure() {
 
     @Override
     protected void containerIsStarting(InspectContainerResponse containerInfo) {
+        List<String> networkAliases = getNetworkAliases();
+        String plaintextAdvertisedListener;
+        if (!getEnvMap().containsKey("KAFKA_BROKER_ID") && listeners.isEmpty() && networkAliases.size() > 1) {
+            // 0 is the random network alias generated by GenericContainer
+            plaintextAdvertisedListener = "PLAINTEXT://" + networkAliases.get(1) + ":" + KafkaHelper.KAFKA_PORT;
+        } else {
+            plaintextAdvertisedListener = "PLAINTEXT://" + getBootstrapServers();
+        }
+
         String brokerAdvertisedListener = String.format(
             "BROKER://%s:%s",
             containerInfo.getConfig().getHostName(),
             "9093"
         );
+
         List<String> advertisedListeners = new ArrayList<>();
-        advertisedListeners.add("PLAINTEXT://" + getBootstrapServers());
+        advertisedListeners.add(plaintextAdvertisedListener);
         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);
@@ -93,8 +102,15 @@ protected void containerIsStarting(InspectContainerResponse containerInfo) {
      * <p>
      * Default advertised listeners:
      * <ul>
-     *      <li>{@code container.getConfig().getHostName():9092}</li>
-     *      <li>{@code container.getHost():container.getMappedPort(9093)}</li>
+     *      <li>{@code container.getConfig().getHostName():container.getMappedPort(9092)}</li>
+     *      <li>{@code container.getHost():9093}</li>
+     * </ul>
+     * <p>
+     * Default advertised listeners if {@code withNetworkAliases} is used, {@code withListener} is not used,
+     * and {@code KAFKA_BROKER_ID} environment variable is not set:
+     * <ul>
+     *      <li>{@code first network alias:9092}</li>
+     *      <li>{@code container.getHost():9093}</li>
      * </ul>
      * @param listener a listener with format {@code host:port}
      * @return this {@link KafkaContainer} instance
@@ -124,8 +140,15 @@ public KafkaContainer withListener(String listener) {
      * <p>
      * Default advertised listeners:
      * <ul>
-     *      <li>{@code container.getConfig().getHostName():9092}</li>
-     *      <li>{@code container.getHost():container.getMappedPort(9093)}</li>
+     *      <li>{@code container.getConfig().getHostName():container.getMappedPort(9092)}</li>
+     *      <li>{@code container.getHost():9093}</li>
+     * </ul>
+     * <p>
+     * Default advertised listeners if {@code withNetworkAliases} is used, {@code withListener} is not used,
+     * and {@code KAFKA_BROKER_ID} environment variable is not set:
+     * <ul>
+     *      <li>{@code first network alias:9092}</li>
+     *      <li>{@code container.getHost():9093}</li>
      * </ul>
      * @param listener a supplier that will provide a listener
      * @param advertisedListener a supplier that will provide a listener

modules/kafka/src/test/java/org/testcontainers/AbstractKafka.java

@@ -1,6 +1,7 @@
 package org.testcontainers;
 
 import com.google.common.collect.ImmutableMap;
+import lombok.SneakyThrows;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.NewTopic;
@@ -15,6 +16,7 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.awaitility.Awaitility;
+import org.testcontainers.containers.Network;
 
 import java.time.Duration;
 import java.util.Collection;
@@ -150,4 +152,16 @@ protected static String getJaasConfig() {
             "user_test=\"secret\";";
         return jaasConfig;
     }
+
+    @SneakyThrows
+    protected void assertKafka(String listener, Network network) {
+        try (KCatContainer kcat = new KCatContainer().withNetwork(network)) {
+            kcat.start();
+
+            kcat.execInContainer("kcat", "-b", listener, "-t", "msgs", "-P", "-l", "/data/msgs.txt");
+            String stdout = kcat.execInContainer("kcat", "-b", listener, "-C", "-t", "msgs", "-c", "1").getStdout();
+
+            assertThat(stdout).contains("Message produced by kcat");
+        }
+    }
 }

modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java

@@ -185,6 +185,19 @@ public void testKraftPrecedenceOverEmbeddedZookeeper() throws Exception {
         }
     }
 
+    @Test
+    public void testUsageWithNetworkAlias() {
+        try (
+            Network network = Network.newNetwork();
+            KafkaContainer kafka = new KafkaContainer(KAFKA_KRAFT_TEST_IMAGE)
+                .withNetworkAliases("mykafka")
+                .withNetwork(network)
+        ) {
+            kafka.start();
+            assertKafka("mykafka:9092", network);
+        }
+    }
+
     @Test
     public void testUsageWithListener() throws Exception {
         try (

modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java

@@ -9,8 +9,6 @@
 import org.testcontainers.containers.SocatContainer;
 import org.testcontainers.utility.MountableFile;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 public class ConfluentKafkaContainerTest extends AbstractKafka {
 
     @Test
@@ -24,6 +22,19 @@ public void testUsage() throws Exception {
         }
     }
 
+    @Test
+    public void testUsageWithNetworkAlias() {
+        try (
+            Network network = Network.newNetwork();
+            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")
+                .withNetworkAliases("mykafka")
+                .withNetwork(network)
+        ) {
+            kafka.start();
+            assertKafka("mykafka:9092", network);
+        }
+    }
+
     @Test
     public void testUsageWithListener() throws Exception {
         try (
@@ -36,14 +47,7 @@ public void testUsageWithListener() throws Exception {
             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");
+            assertKafka("kafka:19092", network);
         }
     }
 

modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java

@@ -2,12 +2,9 @@
 
 import org.junit.Test;
 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;
-
 public class KafkaContainerTest extends AbstractKafka {
 
     @Test
@@ -22,25 +19,32 @@ public void testUsage() throws Exception {
     }
 
     @Test
-    public void testUsageWithListener() throws Exception {
+    public void testUsageWithNetworkAlias() {
+        try (
+            // registerAlias {
+            Network network = Network.newNetwork();
+            KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0")
+                .withNetworkAliases("kafka")
+                .withNetwork(network);
+            // }
+        ) {
+            kafka.start();
+            assertKafka("kafka:9092", network);
+        }
+    }
+
+    @Test
+    public void testUsageWithListener() {
         try (
             Network network = Network.newNetwork();
             // registerListener {
             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");
+            assertKafka("kafka:19092", network);
         }
     }