diff --git a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java
index 61e790d474f..d6bf2305cf7 100644
--- a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java
+++ b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java
@@ -8,6 +8,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
@@ -78,9 +79,13 @@ static void resolveListeners(GenericContainer<?> kafkaContainer, Set<String> lis
             String listenerHost = listener.split(":")[0];
             String listenerPort = listener.split(":")[1];
             String listenerProtocol = String.format("%s://%s:%s", protocol, listenerHost, listenerPort);
-            String protocolMap = String.format("%s:PLAINTEXT", protocol);
             listeners.add(listenerProtocol);
-            listenerSecurityProtocolMap.add(protocolMap);
+            boolean securityProtocolConfigured = listenerSecurityProtocolMap
+                .stream()
+                .anyMatch(entry -> entry.trim().toUpperCase(Locale.ROOT).startsWith(protocol + ":"));
+            if (!securityProtocolConfigured) {
+                listenerSecurityProtocolMap.add(String.format("%s:PLAINTEXT", protocol));
+            }
 
             String host = listener.split(":")[0];
             kafkaContainer.withNetworkAliases(host);
diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java
index 1c95bf95e02..07579b220ea 100644
--- a/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java
+++ b/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java
@@ -66,6 +66,30 @@ void testUsageWithListenerFromProxy() throws Exception {
         }
     }
 
+    @SneakyThrows
+    @Test
+    void shouldKeepSecurityProtocolConfiguredForListener() {
+        try (
+            Network network = Network.newNetwork();
+            SocatContainer socat = new SocatContainer().withNetwork(network).withTarget(2000, "kafka", 19092);
+            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.7.0")
+                .withListener("kafka:19092", () -> socat.getHost() + ":" + socat.getMappedPort(2000))
+                .withNetwork(network)
+                .withEnv(
+                    "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
+                    "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,TC-0:SASL_PLAINTEXT"
+                )
+                .withEnv("KAFKA_LISTENER_NAME_TC-0_SASL_ENABLED_MECHANISMS", "PLAIN")
+                .withEnv("KAFKA_LISTENER_NAME_TC-0_PLAIN_SASL_JAAS_CONFIG", getJaasConfig())
+        ) {
+            socat.start();
+            kafka.start();
+
+            String bootstrapServers = String.format("%s:%s", socat.getHost(), socat.getMappedPort(2000));
+            testSecurePlainKafkaFunctionality(bootstrapServers);
+        }
+    }
+
     @SneakyThrows
     @Test
     void shouldConfigureAuthenticationWithSaslUsingJaas() {