Problem configuring MSK with IAM based authentication in Springboot application

We recently migrated from self-managed Kafka instance to fully-managed AWS MSK cluster. We have only IAM based role-authentication enabled to connect to MSK cluster from local systems.

When I do telnet to the public url of the cluster, I get successful response, but when trying to start my java application, it fails due to different errors. Below is my KafkaConfiguration

Error :

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

@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());
    }
}

Consumer Configuration :

@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;
    }
}