NIfi: javax.security.auth.login.LoginException: No LoginModule found for org.apache.kafka.common.security.plain.PlainLoginModule Error

Hi Team, I am implementing the CSFLE(Client Side Field Level Encryption) in the Nifi for the Kafka data. I have written Groovy script to get the logic like Consume/deserialize/Encrypt the data. The script is working as standalone. However, if i place it in ExecuteScript processor ,i am getting Caused by: javax.security.auth.login.LoginException: No LoginModule found for org.apache.kafka.common.security.plain.PlainLoginModule Error. It seems it is compatibility issue in the kafka-client jar file. However, i could not find the fix. Please provide your input on this.

My groovy script is below.  

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer
import java.time.Duration
import org.apache.kafka.common.security.plain.PlainLoginModule
import javax.security.auth.login.Configuration
import javax.security.auth.login.AppConfigurationEntry
import javax.security.auth.login.LoginContext


Thread.currentThread().setContextClassLoader(null);
def kafkaConsumer = null
try {
    // Kafka consumer properties
    def props = [
        'bootstrap.servers'                    : 'server:port',
        'auto.register.schemas'                : 'false',
        'basic.auth.credentials.source'        : 'USER_INFO',
		'key.serializer'                       : 'org.apache.kafka.common.serialization.StringSerializer',
        'value.serializer'                     : 'io.confluent.kafka.serializers.KafkaAvroSerializer',
        'key.deserializer'                     : 'org.apache.kafka.common.serialization.StringDeserializer',
        'value.deserializer'                   : 'io.confluent.kafka.serializers.KafkaAvroDeserializer',
        'security.protocol'                    : 'SASL_SSL',
        'schema.registry.url'                  : 'Sampe-urlcloud',
        'basic.auth.user.info'                 : 'unamee:pass',
        'sasl.mechanism'                       : 'PLAIN',
        'sasl.jaas.config'                     : 'org.apache.kafka.common.security.plain.PlainLoginModule required username="uname" password="pass";',
        'use.latest.version'                   : 'false',
        'client.id'                            : 'csfle-simple-producer',
        'auto.offset.reset'                    : 'earliest',
        'group.id'                             : 'nifi-csfle-group6',
        'rule.executors._default_.param.access.key.id' :'keyid',
        'rule.executors._default_.param.secret.access.key' : 'accesskey'
    ]
    log.info("RAJIV-1: 'csfle_test_kek_oct_29'")
    // Create KafkaConsumer instance
    kafkaConsumer = new KafkaConsumer<String, Object>(props)
	log.info("RAJIV-2: 'csfle_test_kek_oct_29'")
    kafkaConsumer.subscribe(['csfle_test_kek_oct_29'])
	log.info("RAJIV-3: 'csfle_test_kek_oct_29'")

    // Poll for messages
    ConsumerRecords<String, Object> records = kafkaConsumer.poll(Duration.ofSeconds(10))

    if (records.isEmpty()) {
        log.info("No records found in the topic.")
    } else {
        // Process each record
        records.each { ConsumerRecord<String, Object> record ->
            try {
                // Log the record details
                log.info("Received record: Topic=${record.topic()}, Partition=${record.partition()}, Offset=${record.offset()}, Key=${record.key()}, Value=${record.value()}")
            } catch (Exception e) {
                log.error("Error processing record: ${e.getMessage()}")
            }
        }
    }
} catch (Exception e) {
   log.error("Kafka Consumer Error: ${e.message}", e)

} finally {
    // Close the Kafka consumer
    if (kafkaConsumer != null) {
        kafkaConsumer.close()
    }
}