Forcing PLAINTEXT in SECURITY_PROTOCOL_MAP for created listener

[sfc-gh-krosinski]

I'm trying to create ConfluentKafkaContainer in which I need to have an additional listener. The problem is that I require this listener to use SASL_PLAINTEXT not the plaintext. I see that KafkaHelper pushes PLAINTEXT for this listener into the security protocol map, even if I manually defined it as SASL_PLAINTEXT. Is there a workaround for this, or is there an idea to have this working?

https://github.com/testcontainers/testcontainers-java/blob/6e104da1d6e8a4188451f556e9c7cb4ce8eb3164/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java#L81C11-L81C59

Network network = new Network();
var kafka = new ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.0"))
                .withNetwork(network)
                .withNetworkAliases(HOST_ALIAS)
                .withExposedPorts(9092, 9093, 9095)
                .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
                .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG", "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                        "username=\"admin\" " +
                        "password=\"admin-secret\" " +
                        "user_admin=\"admin-secret\";")
                .withEnv("KAFKA_LISTENER_NAME_TC-0_PLAIN_SASL_JAAS_CONFIG", "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                        "username=\"admin\" " +
                        "password=\"admin-secret\" " +
                        "user_admin=\"admin-secret\";")
                    .withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
                .withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN")
                .withListener("kafka:9095")
                .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT,TC-0:SASL_PLAINTEXT");

[eddumelendez]

can you elaborate on what are you trying to do? I can give an idea but withListener is meant to used for connection between containers and I see you are also exposing the port 9095, so, I would like to understand first.

[kyguy]

I was able to reproduce this issue as well, it appears that PLAINTEXT is forced for any external listeners generated by TestContainers. When KafkaContainer starts, it overwrites the value of the KAFKA_LISTENER_SECURITY_PROTOCOL_MAP env var, adding an extra security protocol map entry for the generated listener, i.e."TC-0", with a security protocol of PLAINTEXT. [1] [2] For example, TestContainers changes this:

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:PLAINTEXT, CONTROLLER:PLAINTEXT,  BROKER:SSL, TC-0:SSL"

to this:

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:PLAINTEXT, CONTROLLER:PLAINTEXT,  BROKER:SSL, TC-0:SSL, TC-0:PLAINTEXT"

Even if "TC-0" is already specified in the original KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, the "TC-0" entry that TestContainers appends to the end of the list takes precedence in Kafka. This makes it impossible to define listeners for connections outside of the container network for any protocols other than PLAINTEXT using the existing TestContainer API.

Workaround

We can only define external listeners for other protocols like SASL_PLAINTEXT and SSLby overriding the configure() [1] method of the KafkaContainer class, this way the resolveListeners() method of the KafkaHelper class [2] doesn't get called and overwrite the user provided listener and listener.security.protocol.map configurations. For example:

var kafka = new ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.0")) {
    @Override
    protected void configure() {
    }
}
.withNetwork(network)
...

Note that we still need to add the external listener names prefixed with "TC-" that TestContainer generates to our KAFKA_LISTENERS and KAFKA_LISTENER_SECURITY_PROTOCOL_MAP env var lists.

Although it is possible to expose external listeners over SSL by overriding methods of the KafkaContainer class, it would be really cool if we could do it through the TestContainer API. I would be happy to contribute an enhancement for it myself if it is of interest to the maintainers and community members.

[1] https://github.com/testcontainers/testcontainers-java/blob/main/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java#L52
[1] https://github.com/testcontainers/testcontainers-java/blob/main/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaHelper.java#L74-L83