TestContainers开展Kafka集成测试时SASL启用机制设置不生效

• 镜像配置转译规则不匹配
  TestContainers 默认对接的 Kafka 容器镜像（Apache 官方、Confluent 官方 Kafka 镜像），本身没有内置「将 KAFKA_ 前缀环境变量自动转写为 broker server.properties 配置项」的逻辑，这套转译规则是 Bitnami 系列 Kafka 镜像独有的启动脚本实现。你硬塞的 KAFKA_SASL_ENABLED_MECHANISMS、KAFKA_SASL_JAAS_CONFIG 这类环境变量，容器启动流程根本不会读取，更不会写入 broker 实际加载的配置文件中，最终 broker 启动时没有任何 SASL 相关配置，自然启用机制列表为空。
• 配置值本身存在错误
  你在 SASL 启用机制配置中混入了非法值：SASL_PLAINTEXT 是监听器的安全协议类型，不是 SASL 认证机制，合法的启用机制值仅包含 PLAIN、SCRAM-SHA-256、SCRAM-SHA-512、GSSAPI 这类认证协议标识。就算环境变量被成功读取，错误值也会导致配置被丢弃。
• TestContainers 默认配置的覆盖逻辑
  KafkaContainer 类在启动前会自动生成完整的 broker server.properties 配置，未纳入配置生成流程的自定义环境变量，不会进入最终生效的配置文件，相当于完全被忽略。

场景1：使用 TestContainers 官方默认的 Kafka 镜像 #

不要硬塞自定义环境变量，直接调用 KafkaContainer 内置的监听器配置 API，它会自动完成 SASL 相关的配置注入、端口映射、advertised 地址设置，完全可以复用你业务侧的 JAAS 配置，不需要重复编写测试代码。示例代码：

// 与业务侧通用的JAAS配置
private static final String JAAS_CONFIG = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
    "username=\"OnlySuperUser\" password=\"super-secret-pass\" " +
    "user_OnlySuperUser=\"super-secret-pass\" " +
    "user_normal_user=\"normal-secret-pass\";";

public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse(KAFKA_TEST_IMAGE))
    .withNetwork(Network.newNetwork())
    .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "false")
    .withEnv("KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "true")
    .withEnv("KAFKA_SUPER_USERS", "User:OnlySuperUser")
    // 内置API配置带SASL认证的监听器，自动完成所有相关配置注入
    .withListener(
        KafkaListenerInstance.saslPlaintext()
            .name("CLIENT_SASL")
            .saslMechanism("PLAIN")
            .jaasConfig(JAAS_CONFIG)
    );

场景2：坚持使用硬塞环境变量的方式（适配 Bitnami Kafka 镜像） #

必须严格匹配 Bitnami 镜像的环境变量命名规则，修正之前的配置错误：

• 所有 broker 配置项统一使用 KAFKA_CFG_ 为前缀（而非无规则的 KAFKA_ 前缀）
• 修正 SASL 机制配置值，去掉非法的 SASL_PLAINTEXT
• 显式配置监听器与安全协议的映射关系，明确指定 SASL_PLAINTEXT 为监听器的安全协议

示例配置：

public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("bitnami/kafka:3.7"))
    .withNetwork(Network.newNetwork())
    .withEnv("KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE", "false")
    .withEnv("KAFKA_CFG_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "true")
    .withEnv("KAFKA_CFG_SUPER_USERS", "User:OnlySuperUser")
    .withEnv("KAFKA_CFG_SASL_ENABLED_MECHANISMS", "PLAIN")
    .withEnv("KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN")
    .withEnv("KAFKA_CFG_SASL_JAAS_CONFIG", JAAS_CONFIG)
    .withEnv("KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP", "CLIENT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT")
    .withEnv("KAFKA_CFG_LISTENERS", "CLIENT://:9092,CONTROLLER://:9093");

内容的提问来源于stack exchange，提问作者Ryan