Problem configuring MSK with IAM based authentication in Springboot application

0

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;
    }
}
Keine Antworten

Du bist nicht angemeldet. Anmelden um eine Antwort zu veröffentlichen.

Eine gute Antwort beantwortet die Frage klar, gibt konstruktives Feedback und fördert die berufliche Weiterentwicklung des Fragenstellers.

Richtlinien für die Beantwortung von Fragen