docs and confluent

Author: julianladisch

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. 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.
+
 ## Options
         
 ### <a name="zookeeper"></a> Using external Zookeeper

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

@@ -48,13 +48,24 @@ protected void configure() {
 
     @Override
     protected void containerIsStarting(InspectContainerResponse containerInfo) {
+        List<String> networkAliases = getNetworkAliases();
+        String plaintextAdvertisedListener;
+        if (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 +98,14 @@ 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 withListener} is not used but {@code withNetworkAliases} is used:
+     * <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 +135,14 @@ 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 withListener} is not used but {@code withNetworkAliases} is used:
+     * <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,23 +54,25 @@ protected void configure() {
 
     @Override
     protected void containerIsStarting(InspectContainerResponse containerInfo) {
-        String brokerAdvertisedListener = String.format(
-            "BROKER://%s:%s",
-            containerInfo.getConfig().getHostName(),
-            "9093"
-        );
-
         List<String> networkAliases = getNetworkAliases();
-        String plaintextServer;
+        String plaintextAdvertisedListener;
         if (listeners.isEmpty() && networkAliases.size() > 1) {
             // 0 is the random network alias generated by GenericContainer
-            plaintextServer = networkAliases.get(1) + ":" + KAFKA_PORT;
+            plaintextAdvertisedListener =
+                    "PLAINTEXT://" + networkAliases.get(1) + ":" + KafkaHelper.KAFKA_PORT;
         } else {
-            plaintextServer = getBootstrapServers();
+            plaintextAdvertisedListener =
+                    "PLAINTEXT://" + getBootstrapServers();
         }
 
+        String brokerAdvertisedListener = String.format(
+            "BROKER://%s:%s",
+            containerInfo.getConfig().getHostName(),
+            "9093"
+        );
+
         List<String> advertisedListeners = new ArrayList<>();
-        advertisedListeners.add("PLAINTEXT://" + plaintextServer);
+        advertisedListeners.add(plaintextAdvertisedListener);
         advertisedListeners.add(brokerAdvertisedListener);
 
         advertisedListeners.addAll(KafkaHelper.resolveAdvertisedListeners(this.advertisedListeners));
@@ -102,8 +104,14 @@ 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 withListener} is not used but {@code withNetworkAliases} is used:
+     * <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
@@ -133,8 +141,14 @@ 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 withListener} is not used but {@code withNetworkAliases} is used:
+     * <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