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()
}
}
... View more
