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 209534fd3db..870ab03bd0e 100644 --- a/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java +++ b/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java @@ -323,14 +323,8 @@ void withClusterId(String clusterId) { void withRaft() { this.envVars.computeIfAbsent("CLUSTER_ID", key -> clusterId); this.envVars.computeIfAbsent("KAFKA_NODE_ID", key -> getEnvVars().get("KAFKA_BROKER_ID")); - addEnvVar( - "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", - String.format("%s,CONTROLLER:PLAINTEXT", getEnvVars().get("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP")) - ); - addEnvVar( - "KAFKA_LISTENERS", - String.format("%s,CONTROLLER://0.0.0.0:9094", getEnvVars().get("KAFKA_LISTENERS")) - ); + addEnvVar("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", kafkaListenerSecurityProtocolMap()); + addEnvVar("KAFKA_LISTENERS", kafkaListeners()); addEnvVar("KAFKA_PROCESS_ROLES", "broker,controller"); String firstNetworkAlias = getNetworkAliases().stream().findFirst().orElse(null); @@ -345,5 +339,24 @@ void withRaft() { setWaitStrategy(Wait.forLogMessage(".*Transitioning from RECOVERY to RUNNING.*", 1)); } + + private String kafkaListenerSecurityProtocolMap() { + String kafkaListenerSecurityProtocolMapEnvVar = getEnvVars().get("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"); + String kafkaListenerSecurityProtocolMap = String.format( + "%s,CONTROLLER:PLAINTEXT", + kafkaListenerSecurityProtocolMapEnvVar + ); + Set listenerSecurityProtocolMap = new HashSet<>( + Arrays.asList(kafkaListenerSecurityProtocolMap.split(",")) + ); + return String.join(",", listenerSecurityProtocolMap); + } + + private String kafkaListeners() { + String kafkaListenersEnvVar = getEnvVars().get("KAFKA_LISTENERS"); + String kafkaListeners = String.format("%s,CONTROLLER://0.0.0.0:9094", kafkaListenersEnvVar); + Set listeners = new HashSet<>(Arrays.asList(kafkaListeners.split(","))); + return String.join(",", listeners); + } } }