Testcontainers-Java项目中Kafka容器认证配置问题解析

问题背景

在使用Testcontainers-Java项目进行Kafka集成测试时，开发者Valery-Pechur遇到了容器认证配置问题。该问题出现在尝试为Kafka容器配置SCRAM-SHA-256认证机制时，客户端连接时出现了认证失败的情况。

初始配置分析

开发者最初尝试了以下配置方案：

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)
);

这种配置方式使用了Zookeeper模式，并通过kafka-configs命令添加了SCRAM用户凭据。这种方式在旧版本中工作正常，但在迁移到新版本时出现了问题。

新版本配置尝试

开发者随后尝试了新的配置方案，针对Kafka的KRaft模式(无Zookeeper)进行了调整：

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))
);

这种配置尝试为不同的监听器配置不同的认证机制，但出现了错误："Unexpected handshake request with client mechanism SCRAM-SHA-256, enabled mechanisms are []"，表明客户端尝试使用SCRAM-SHA-256机制，但服务端没有启用该机制。

问题原因分析

1. KRaft模式配置差异：从Zookeeper模式迁移到KRaft模式时，用户凭据的添加方式发生了变化。在KRaft模式下，需要使用kafka-storage命令而不是kafka-configs命令来添加用户。

2. 监听器配置问题：初始配置中可能没有正确地将SCRAM-SHA-256机制应用到正确的监听器上，导致服务端没有启用预期的认证机制。

3. 环境变量命名规范：不同版本的Kafka镜像可能对环境变量的命名有不同的要求，如使用KAFKA_CFG_前缀还是直接使用KAFKA_前缀。

解决方案探索

开发者随后尝试了第三种配置方案：

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))
);

这种配置简化了监听器设置，专注于PLAINTEXT监听器的SCRAM-SHA-256认证。虽然错误消息变成了"Authentication failed during authentication due to invalid credentials"，表明认证机制已经启用，但凭据验证失败，这说明配置已经部分生效。

最佳实践建议

1. 明确Kafka运行模式：确定使用Zookeeper模式还是KRaft模式，因为两者的配置方式不同。

2. 简化监听器配置：除非有特殊需求，否则可以简化监听器配置，专注于一个主要的SASL监听器。

3. 正确添加用户凭据：

   • 对于Zookeeper模式：使用kafka-configs命令
   • 对于KRaft模式：使用kafka-storage命令

4. 验证环境变量前缀：根据使用的Kafka镜像版本，确认是使用KAFKA_还是KAFKA_CFG_前缀。

5. 分步调试：先确保基础连接正常工作，再逐步添加认证配置。

完整解决方案示例

以下是一个针对KRaft模式的完整配置示例：

private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION = Map.of(
    "KAFKA_CFG_PROCESS_ROLES", "broker,controller",
    "KAFKA_CFG_NODE_ID", "1",
    "KAFKA_CFG_CONTROLLER_QUORUM_VOTERS", "1@localhost:9093",
    "KAFKA_CFG_LISTENERS", "PLAINTEXT://:9092,CONTROLLER://:9093",
    "KAFKA_CFG_ADVERTISED_LISTENERS", "PLAINTEXT://localhost:9092",
    "KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP", "CONTROLLER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT",
    "KAFKA_CFG_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256",
    "KAFKA_CFG_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256",
    "KAFKA_CFG_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)
);

// 在容器启动后执行
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"
);

总结

在Testcontainers-Java项目中配置Kafka容器认证时，需要特别注意Kafka的运行模式(Zookeeper或KRaft)以及对应版本的环境变量命名规范。正确的配置顺序应该是：先设置容器环境变量定义认证机制，然后启动容器，最后执行命令添加用户凭据。通过分步调试和简化配置，可以有效地解决认证配置问题。