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

Question

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.

Answer 1

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