How to setup SASL_SSL with parallel kafka consumer

[rock89]

I am able to consume message with kafka broker without SASL_SSL mechanism enabled but not able to connect when SASL_SSL mechanism is enabled with brokers , getting timeout issue .

DEBUG o.a.k.c.NetworkClient - [ - ] - [] - [Producer clientId=ingestion-service-1] Initiating connection to node localhost:9096 (id: -2 rack: null) using address localhost:9096/<ip>
2023-07-25 09:40:09,892 [pc-broker-poll] DEBUG o.a.k.c.n.Selector - [ - ] - [] - [Consumer clientId=service-1, groupId=service-1] Connection with localhost:9096/<IP> disconnected
! java.io.EOFException: null
! at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:120)
! at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:470)
! at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:560)
! at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:248)
! at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:176)
! at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)

broker kafka version : 2.8.0

<dependency>
            <groupId>io.confluent.parallelconsumer</groupId>
            <artifactId>parallel-consumer-core</artifactId>
            <version>0.5.2.4</version>
        </dependency>

Kafka Client setup:

private Map<String, Object> createBaseKafkaConfigurations() {
       Map<String, Object> config = new HashMap();
       config.put("bootstrap.servers", String.join(",", this.kafkaConfiguration.getBootstrapServers()));
       if (!this.topicConfiguration.isIgnoreSchemaRegistry()) {
           Optional.ofNullable(this.kafkaConfiguration.getSchemaRegistryServers()).ifPresent((strings) -> {
               config.put("schema.registry.url", String.join(",", strings));
           });
       }

       config.put("client.id", this.kafkaConfiguration.getClientId());

       config.put("group.id", this.topicConfiguration.getGroupId());
       config.put("key.deserializer", this.topicConfiguration.getKeyDeserializerClass());
       config.put("value.deserializer", this.topicConfiguration.getValueDeserializerClass());
       config.put("max.poll.records", this.topicConfiguration.getMaxPollRecords());
       config.put("auto.offset.reset", this.topicConfiguration.getAutoOffsetReset());
       config.put("enable.auto.commit", this.topicConfiguration.isAutoCommitEnabled());
       Optional.ofNullable(this.topicConfiguration.getMaxPollIntervalMs()).ifPresent((maxPollInterval) -> {
           config.put("max.poll.interval.ms", maxPollInterval);
       });

       config.put("security.protocol", "SASL_SSL");
       config.put("sasl.mechanism", "SCRAM-SHA-512");
       config.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"${SERVICE_SASL_USER_NAME}\" password=\"${SERVICE_SASL_PASSWORD}\";");



       if (this.topicConfiguration.isAutoCommitEnabled()) {
           config.put("auto.commit.interval.ms", this.topicConfiguration.getAutoCommitInterval());
       }

       return config;
   }

ParallelStreamProcessor<K, V> setupParallelConsumer(Map<String, Object> config) {
       // tag::exampleSetup[]
       Consumer<K, V> kafkaConsumer = buildConsumer(config); // <1>

       var options = ParallelConsumerOptions.<K, V>builder()
                                            .ordering(KEY) // <2>
                                            .maxConcurrency(16) // <3>
                                            .consumer(kafkaConsumer)
                                            .build();

       return ParallelStreamProcessor.createEosStreamProcessor(options);

[rock89]

Got the issue.