How to configure ConfluentKafkaContainer with SASL/SCRAM

[Valery-Pechur]

Module

Kafka

Testcontainers version

1.20.4

Using the latest Testcontainers version?

Yes

Host OS

Windows

Host Arch

x64

Docker version

Docker version 27.0.3

What happened?

Authentication error when starting test container

@UtilityClass
@Slf4j
public class KafkaInitializer {

    private static final String EMP_KAFKA_BOOTSTRAP_SERVERS_PROPERTY_NAME = "kafka.bootstrap-servers=";
    private static final String AUTH_USERNAME = "kafka-test";
    private static final String AUTH_PASSWORD = "kafka-test-password";

    private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION =  Map.ofEntries(

            Map.entry("KAFKA_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256,PLAIN"),
            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256"),
            Map.entry("KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256"),
            Map.entry("KAFKA_LISTENER_NAME_CONTROLLER_SASL_ENABLED_MECHANISMS", "PLAIN"),
            Map.entry("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "SCRAM-SHA-256"),
            Map.entry("KAFKA_SASL_MECHANISM_CONTROLLER_PROTOCOL", "PLAIN"),

            Map.entry("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"),

            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SCRAM-SHA-256_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD)),

            Map.entry("KAFKA_LISTENER_NAME_BROKER_SCRAM-SHA-256_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD)),

            Map.entry("KAFKA_LISTENER_NAME_CONTROLLER_PLAIN_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.plain.PlainLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD))
    );

    @Container
    private static final ConfluentKafkaContainer KAFKA_CONTAINER = new ConfluentKafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.7.0")
    );

    public static class Initializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {

        @Override
        public void initialize(@NotNull ConfigurableApplicationContext applicationContext) {
            try {
           
                KAFKA_CONTAINER.withEnv(KAFKA_ENV_WITH_AUTHORIZATION)
                        .start();


            KAFKA_CONTAINER.execInContainer(
                        "/usr/bin/kafka-storage", "format", "-t", " $CLUSTER_ID",
                        "--config", "/etc/kafka/kraft/server.properties",
                        "--add-scram", "SCRAM-SHA-256=[name=%s,password=%s]".formatted(AUTH_USERNAME, AUTH_PASSWORD),
                        "--ignore-formatted"
                );

                log.info("KafkaContainer for tests started");
            } catch (Throwable throwable) {
                log.error("KafkaContainer not started: {}", throwable.getMessage());
            }

            try {
                String propertyBootstrapServers = EMP_KAFKA_BOOTSTRAP_SERVERS_PROPERTY_NAME + KAFKA_CONTAINER.getBootstrapServers();
                TestPropertyValues.of(
                        propertyBootstrapServers
                ).applyTo(applicationContext.getEnvironment());
                log.info("TestPropertyValues of KafkaContainer: propertyBootstrapServers = {} was applied", propertyBootstrapServers);
            } catch (Throwable throwable) {
                log.error(
                        "TestPropertyValues of KafkaContainer don't apply for applicationContext: {}",
                        throwable.getMessage()
                );
                throw throwable;
            }
        }
    }
}

Additionally, the following settings are set:

  if (authUsername != null && authPassword != null) {
            result.put(SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
            result.put(SASL_MECHANISM, "SCRAM-SHA-256");
            result.put(SASL_JAAS_CONFIG, String.format(SASL_JAAS_CONFIG_TEMPLATE, authUsername, authPassword));
        }

These settings are set in DefaultKafkaConsumerFactory<>(consumerProperties, new StringDeserializer(), deserializer); and in ProducerFactory

Relevant log output

[ERROR] [Consumer clientId=consumer-user-role-test-1, groupId=user-role-test] Connection to node -1 (localhost/127.0.0.1:51382) failed authentication due to: Unexpected handshake request with client mechanism SCRAM-SHA-256, enabled mechanisms are []
[ERROR] Fatal consumer exception; stopping container
[ERROR] [Producer clientId=producer-1] Connection to node -1 (localhost/127.0.0.1:51382) failed authentication due to: Unexpected handshake request with client mechanism SCRAM-SHA-256, enabled mechanisms are []

Additional Information

Before changing the container, the following code worked:

 private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION = Map.of(
            "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT",
            "KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", ScramMechanism.SCRAM_SHA_256.mechanismName(),
            "KAFKA_LISTENER_NAME_PLAINTEXT_SCRAM-SHA-256_SASL_JAAS_CONFIG",
            "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                    .formatted(AUTH_USERNAME, AUTH_PASSWORD)
    );

   @Container
    private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.7.0")
    );

    public static class Initializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {

        @Override
        public void initialize(@NotNull ConfigurableApplicationContext applicationContext) {
            try {
      
                KAFKA_CONTAINER.withEnv(KAFKA_ENV_WITH_AUTHORIZATION).start();

                KAFKA_CONTAINER.execInContainer(
                        "kafka-configs", "--zookeeper", "localhost:2181", "--alter",
                        "--add-config", "SCRAM-SHA-256=[iterations=4096,password=%s]".formatted(AUTH_PASSWORD),
                        "--entity-type", "users", "--entity-name", AUTH_USERNAME
                );
                log.info("KafkaContainer for tests started");

[Valery-Pechur]

After changed the settings to:

  private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION =  Map.ofEntries(

            Map.entry("KAFKA_CFG_SASL_ENABLED_MECHANISMS", "[SCRAM-SHA-256]"),
            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256"),

            Map.entry("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT,BROKER:PLAINTEXT"),

            Map.entry("KAFKA_CLIENT_LISTENER_NAME", "PLAINTEXT"),

            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SCRAM-SHA-256_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD))
    );

I have next error:

[ERROR] [Consumer clientId=consumer-user-role-test-1, groupId=user-role-test] Connection to node -1 (localhost/127.0.0.1:50171) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256
[ERROR] Fatal consumer exception; stopping container

[eddumelendez]

I have submitted a couple of examples to configure KafkaContainer and ConfluentKafkaContainer with SASL/SCRAM auth. See 92a475d

[Valery-Pechur]

Try replacing KAFKA_CFG_SASL_ENABLED_MECHANISMS with KAFKA_SASL_ENABLED_MECHANISMS

KAFKA_SASL_ENABLED_MECHANISMS

[Valery-Pechur]

zookeeper

I know that. org.testcontainers.containers.KafkaContainer is marked as deprecated. when upgrading to ConfluentKafkaContainer zookeeper was changed to Kraft and authorization for Kraft mode needs to be configured

[eddumelendez]

Please, LMK if the examples mentioned above help

[lazy93-design]

Please, LMK if the examples mentioned above help

Hi,
I've just used your examples (from tests) for starting kafka container with scram - it works, thanks!