apache-kafkaquarkusjaassmallrye-reactive-messaging

Quarkus: Smallrye Kafka configure channels for distinct bootstrap servers using diferent KEYS and PASSWORD


My goal is to produce events in 2 different channels using distinct bootstrap servers using jaas configuration using SASL_SSL, but I am not able to setup the channels to authenticate correctly on the bootstrap servers.

I've tried the following setup

mp.messaging.outgoing.channel1.bootstrap.servers=${KAFKA1}
mp.messaging.outgoing.channel1.ssl.endpoint-identification-algorithm=https
mp.messaging.outgoing.channel1.security.protocol=SASL_SSL
mp.messaging.outgoing.channel1.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${KEY1}" password="${PWD1}";
mp.messaging.outgoing.channel1.sasl.mechanism=PLAIN

mp.messaging.outgoing.channel2.bootstrap.servers=${KAFKA2}
mp.messaging.outgoing.channel2.ssl.endpoint-identification-algorithm=https
mp.messaging.outgoing.channel2.security.protocol=SASL_SSL
mp.messaging.outgoing.channel2.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${KEY2}" password="${PWD2}";
mp.messaging.outgoing.channel2.sasl.mechanism=PLAIN

Using this setup I am receiving errors on the channel initialization.

2023-01-18 13:57:10 13:57:10.445 ERROR [Application] (main) Failed to start application (with profile prod): java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set
2023-01-18 13:57:10     at org.apache.kafka.common.security.JaasContext.defaultContext(JaasContext.java:131)
2023-01-18 13:57:10     at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:96)
2023-01-18 13:57:10     at org.apache.kafka.common.security.JaasContext.loadClientContext(JaasContext.java:82)
2023-01-18 13:57:10     at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:167)
2023-01-18 13:57:10     at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81)
2023-01-18 13:57:10     at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105)

The initial setup used the default bootstrap settings and it worked fined until KAFKA was brought to the equation.

kafka.bootstrap.servers='${KAFKA1}'
kafka.ssl.endpoint-identification-algorithm=https
kafka.security.protocol=SASL_SSL
kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${Key1}" password="${PWD1}";
kafka.sasl.mechanism=PLAIN

I've tried the described on the issue and I am not being able figure out how to configure the channels to authenticate to 2 different bootstrap servers.


Solution

  • After reading the documentation [https://quarkus.io/guides/kafka] the option 1

    kafka-configuration Allows the config1

    Therefore the solution is to implement a provider bean

    @ApplicationScoped
    @Slf4j
    public class KafkaConfigBean {
    
        @Produces
        @Identifier("kafka1")
        public Map<String, Object> kafkaConfig() {
            HashMap<String, Object> config = new HashMap<>();
    
            config.put("security.protocol", "SASL_SSL");
            config.put("sasl.mechanism", "PLAIN");
    
            String saslConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%S\" password=\"%s\";",
                    System.getenv("KEY1"), System.getenv("PWD1"));
            config.put("sasl.jaas.config", saslConfig);
    
            log.info("Initialized Kafka 1 config");
    
            return config;
        }
    
        @Produces
        @Identifier("kafka2")
        public Map<String, Object> kafkaConfigPTT() {
            HashMap<String, Object> config = new HashMap<>();
    
            config.put("security.protocol", "SASL_SSL");
            config.put("sasl.mechanism", "PLAIN");
    
            String saslConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%S\" password=\"%s\";",
                    System.getenv("KEY2"), System.getenv("PWD2"));
            config.put("sasl.jaas.config", saslConfig);
    
            log.info("Initialized Kafka 2 config");
    
            return config;
        }
    }
    

    Thus resulting on the following configuration file

    mp.messaging.outgoing.channel1.bootstrap.servers=\${KAFKA1}
    mp.messaging.outgoing.channel1.kafka-configuration=kafka1
     
    mp.messaging.outgoing.channel2.bootstrap.servers=\${KAFKA2}
    mp.messaging.outgoing.channel2.kafka-configuration=kafka2