Additional listeners should inherit the configured authentication method (#7594)

Co-authored-by: Eddú Meléndez Gonzales <[email protected]>

Author: lburgazzoli

modules/redpanda/src/main/java/org/testcontainers/redpanda/RedpandaContainer.java

@@ -232,6 +232,7 @@ private Transferable getRedpandaFile(Configuration cfg) {
                     Map<String, Object> listenerMap = new HashMap<>();
                     listenerMap.put("address", listener.getAddress());
                     listenerMap.put("port", listener.getPort());
+                    listenerMap.put("authentication_method", this.authenticationMethod);
                     return listenerMap;
                 })
                 .collect(Collectors.toList());

modules/redpanda/src/main/resources/testcontainers/redpanda.yaml.ftl

@@ -25,6 +25,7 @@ redpanda:
     - address: 0.0.0.0
       name: ${listener.address}
       port: ${listener.port}
+      authentication_method: ${listener.authentication_method}
 </#list>
 
   advertised_kafka_api:

modules/redpanda/src/test/java/org/testcontainers/redpanda/RedpandaContainerTest.java

@@ -31,6 +31,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -138,6 +139,84 @@ public void testUsageWithListener() throws Exception {
         }
     }
 
+    @Test
+    public void testUsageWithListenerAndSasl() throws Exception {
+        final String username = "panda";
+        final String password = "pandapass";
+        final String algorithm = "SCRAM-SHA-256";
+
+        try (
+            Network network = Network.newNetwork();
+            RedpandaContainer redpanda = new RedpandaContainer("docker.redpanda.com/redpandadata/redpanda:v23.1.7")
+                .enableAuthorization()
+                .enableSasl()
+                .withSuperuser("panda")
+                .withListener(() -> "my-panda:29092")
+                .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")
+        ) {
+            redpanda.start();
+
+            String adminUrl = String.format("%s/v1/security/users", redpanda.getAdminAddress());
+            Map<String, String> params = new HashMap<>();
+            params.put("username", username);
+            params.put("password", password);
+            params.put("algorithm", algorithm);
+
+            RestAssured.given().contentType("application/json").body(params).post(adminUrl).then().statusCode(200);
+
+            kcat.start();
+
+            kcat.execInContainer(
+                "kcat",
+                "-b",
+                "my-panda:29092",
+                "-X",
+                "security.protocol=SASL_PLAINTEXT",
+                "-X",
+                "sasl.mechanisms=" + algorithm,
+                "-X",
+                "sasl.username=" + username,
+                "-X",
+                "sasl.password=" + password,
+                "-t",
+                "msgs",
+                "-P",
+                "-l",
+                "/data/msgs.txt"
+            );
+
+            String stdout = kcat
+                .execInContainer(
+                    "kcat",
+                    "-b",
+                    "my-panda:29092",
+                    "-X",
+                    "security.protocol=SASL_PLAINTEXT",
+                    "-X",
+                    "sasl.mechanisms=" + algorithm,
+                    "-X",
+                    "sasl.username=" + username,
+                    "-X",
+                    "sasl.password=" + password,
+                    "-C",
+                    "-t",
+                    "msgs",
+                    "-c",
+                    "1"
+                )
+                .getStdout();
+
+            assertThat(stdout).contains("Message produced by kcat");
+        }
+    }
+
     @SneakyThrows
     @Test
     public void enableSaslWithSuccessfulTopicCreation() {