Quarkus: Smallrye Kafka configure channels for distinct bootstrap servers using different KEY and PASSWORD

[hugo-l-pais]

Hi all,

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.

My current config is the one on the code block and it works fine with KAFKA1 cluster

mp.messaging.outgoing.channel1.bootstrap.servers=${KAFKA1}
mp.messaging.outgoing.channel1.security.protocol=SASL_SSL
mp.messaging.outgoing.channel1.sasl.mechanism=PLAIN

mp.messaging.outgoing.channel2.bootstrap.servers=${KAFKA2}
mp.messaging.outgoing.channel2.security.protocol=SASL_SSL
mp.messaging.outgoing.channel2.sasl.mechanism=PLAIN

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

note: using Quarkus 2.15.3.Final

How to configure the authentication for KAFKA2 cluster?

[quarkus-bot[bot]]

/cc @Ladicek (smallrye), @alesj (kafka), @cescoffier (kafka), @jmartisk (smallrye), @ozangunalp (kafka), @phillip-kruger (smallrye), @radcortez (smallrye)

[ozangunalp]

Hi, you can continue using the global kafka config like kafka.sasl.mechanism for common config properties and use per channel ones like mp.messaging.outgoing.channel2.sasl.jaas.config for distinct credentials.

[hugo-l-pais]

Hi @ozangunalp thanks for your help.

I have applied the suggested

mp.messaging.outgoing.channel1.bootstrap.servers=${KAFKA1}
mp.messaging.outgoing.channel1.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${Key1}" password="${PWD1}";
mp.messaging.outgoing.channel1.security.protocol=SASL_SSL

mp.messaging.outgoing.channel2.bootstrap.servers=${KAFKA2}
mp.messaging.outgoing.channel2.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${Key2}" password="${PWD2}";
mp.messaging.outgoing.channel2.security.protocol=SASL_SSL

kafka.ssl.endpoint-identification-algorithm=https
kafka.security.protocol=SASL_SSL
kafka.sasl.mechanism=PLAIN

When deploying the application I am getting the following error now

Am I doing something incorrect? One additional note I have one consumer for KAFKA1 and it has authenticated correctly only the producers are failing.

[hugo-l-pais]

So after a lot of digging I've understood how to configure, as the documentation states a bean can provide additional parameters via kafka-configuration parameter.

The working configuration is the following

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

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