如何通过IAM认证连接SpringBoot微服务到AWS MSK？认证报错求助

错误日志核心是AWS IAM凭证加载失败，你的配置存在两个关键问题：

1. JAAS配置中指定了awsProfileName="ATKAFKA"，但本地AWS配置文件中不存在该profile
2. 额外添加的sasl.aws.accessKeyId和sasl.aws.secretAccessKey并非AWS MSK IAM认证的标准配置项，IAMLoginModule不会读取这两个参数

根据你的部署环境，选择以下一种正确配置方式：

方式1：测试环境 - 硬编码凭证 #

修改JAAS配置，直接传入accessKey和secretKey，移除无效配置项：

@Bean
public KafkaAdmin kafkaAdmin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    configs.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
    // 直接在JAAS中指定凭证
    configs.put(SaslConfigs.SASL_JAAS_CONFIG, String.format(
        "software.amazon.msk.auth.iam.IAMLoginModule required awsAccessKeyId=\"%s\" awsSecretAccessKey=\"%s\" awsStsRegion=\"us-east-1\";",
        accessKey, secretKey
    ));
    configs.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
    
    configs.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
    configs.put(AdminClientConfig.RETRIES_CONFIG, 5);
    // 移除无效配置
    // configs.put("sasl.aws.accessKeyId", accessKey);
    // configs.put("sasl.aws.secretAccessKey", secretKey);
    return new KafkaAdmin(configs);
}

方式2：开发环境 - 本地AWS配置文件 #

1. 确保本地~/.aws/credentials文件中存在ATKAFKA profile：

[ATKAFKA]
aws_access_key_id = 你的AccessKey
aws_secret_access_key = 你的SecretKey

2. 简化Kafka配置，移除硬编码凭证：

@Bean
public KafkaAdmin kafkaAdmin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    configs.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
    configs.put(SaslConfigs.SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required awsProfileName=\"ATKAFKA\" awsStsRegion=\"us-east-1\";");
    configs.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
    
    configs.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
    configs.put(AdminClientConfig.RETRIES_CONFIG, 5);
    // 移除无效配置
    // configs.put("sasl.aws.accessKeyId", accessKey);
    // configs.put("sasl.aws.secretAccessKey", secretKey);
    return new KafkaAdmin(configs);
}

方式3：生产环境 - AWS默认凭证链（推荐） #

在EC2/EKS/ECS等AWS托管环境中，无需硬编码凭证，依赖IAM角色自动加载：

@Bean
public KafkaAdmin kafkaAdmin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    configs.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
    // 无需指定凭证，LoginModule自动读取默认凭证链
    configs.put(SaslConfigs.SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required awsStsRegion=\"us-east-1\";");
    configs.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
    
    configs.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
    configs.put(AdminClientConfig.RETRIES_CONFIG, 5);
    return new KafkaAdmin(configs);
}

同时确保运行环境的IAM角色拥有MSK访问权限（如kafka-cluster:Connect、kafka-cluster:AlterCluster等）。

• 不要同时混用多种凭证配置方式，避免加载逻辑冲突
• 生产环境禁止硬编码凭证，优先使用IAM角色或环境变量
• 确保awsStsRegion与MSK集群所在区域一致

内容的提问来源于stack exchange，提问作者Juan Sebastian Cabra Leon