迁移至AWS MSK后Spring应用IAM认证失败：JAAS配置错误排查

问题场景 #

我们将自建Kafka实例迁移至AWS MSK全托管集群，采用IAM角色认证从本地系统连接集群。telnet测试集群公网地址连通正常，但Java应用启动失败，报错：

Invalid login module control flag 'com.amazonaws.auth.AWSStaticCredentialsProvider' in JAAS config

当前配置代码 #

生产者与Admin配置 #

@Configuration
public class KafkaConfiguration {

    @Value("${aws.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${aws.kafka.accessKey}")
    private String accessKey;

    @Value("${aws.kafka.secret}")
    private String secret;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        AWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secret);
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        configs.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        configs.put(SaslConfigs.SASL_JAAS_CONFIG, "com.amazonaws.auth.AWSCredentialsProvider com.amazonaws.auth.AWSStaticCredentialsProvider(" + awsCredentials + ")");
        return new KafkaAdmin(configs);
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        AWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secret);

        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put("security.protocol", "SASL_SSL");
        configProps.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        configProps.put(SaslConfigs.SASL_JAAS_CONFIG, "com.amazonaws.auth.AWSCredentialsProvider com.amazonaws.auth.AWSStaticCredentialsProvider(" + awsCredentials + ")");
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

消费者配置 #

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${aws.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${aws.kafka.accessKey}")
    private String accessKey;

    @Value("${aws.kafka.secret}")
    private String secret;

    public ConsumerFactory<String, String> consumerFactory() {
        AWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secret);

        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put("security.protocol", "SASL_SSL");
        configProps.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        configProps.put(SaslConfigs.SASL_JAAS_CONFIG, "com.amazonaws.auth.AWSCredentialsProvider com.amazonaws.auth.AWSStaticCredentialsProvider(" + awsCredentials + ")");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "iTopLight");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> rawKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

错误原因与解决方案 #

错误根源 #

JAAS配置格式完全错误。AWS MSK IAM基于OAuthBearer机制，对应的JAAS登录模块是org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule，而非直接引用AWS凭证提供者类。原配置把凭证提供者类当作了登录模块控制标志，导致语法解析失败。

修正后的配置 #

生产者与Admin配置修正

@Configuration
public class KafkaConfiguration {

    @Value("${aws.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${aws.kafka.accessKey}")
    private String accessKey;

    @Value("${aws.kafka.secret}")
    private String secret;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        configs.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        // 修正JAAS配置格式
        configs.put(SaslConfigs.SASL_JAAS_CONFIG, 
            "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
            "aws.credentials.provider.class=com.amazonaws.auth.AWSStaticCredentialsProvider " +
            "aws.access.key.id=\"" + accessKey + "\" " +
            "aws.secret.access.key=\"" + secret + "\";");
        return new KafkaAdmin(configs);
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put("security.protocol", "SASL_SSL");
        configProps.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        // 修正JAAS配置格式
        configProps.put(SaslConfigs.SASL_JAAS_CONFIG, 
            "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
            "aws.credentials.provider.class=com.amazonaws.auth.AWSStaticCredentialsProvider " +
            "aws.access.key.id=\"" + accessKey + "\" " +
            "aws.secret.access.key=\"" + secret + "\";");
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

消费者配置修正

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${aws.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${aws.kafka.accessKey}")
    private String accessKey;

    @Value("${aws.kafka.secret}")
    private String secret;

    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put("security.protocol", "SASL_SSL");
        configProps.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
        // 修正JAAS配置格式
        configProps.put(SaslConfigs.SASL_JAAS_CONFIG, 
            "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
            "aws.credentials.provider.class=com.amazonaws.auth.AWSStaticCredentialsProvider " +
            "aws.access.key.id=\"" + accessKey + "\" " +
            "aws.secret.access.key=\"" + secret + "\";");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "iTopLight");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> rawKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

额外说明 #

1. 无需手动创建BasicAWSCredentials实例，JAAS配置会通过指定的提供者类自动加载凭证
2. 生产环境更推荐使用默认凭证链（比如环境变量、ECS任务角色、EC2实例角色等），此时JAAS配置可简化为：

   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required aws.credentials.provider.class=com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
   

3. 确保项目依赖包含aws-msk-iam-auth库，版本需与Kafka客户端版本兼容

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