Hi Team, I am trying to develop a flow to update the Parameter context on the fly using the InvokeHTTP Processor. Based on the screenshot i have configured the same. However i am getting unAuthorized Error.    My URL format is like : https://<nifi-host>:<port>/nifi-api/parameter-contexts/<context-id> Error Screenshot.     ... View more

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

Hi Team, I am trying to implement for the scenario where the events should sleep for 5 minutes and let all the flow files are queued up, then check the count of the files if it is greater than 10 route to failure else route to SUCCESS. I did in Execute script processor as below. However, i am trying to avoid the executescript for this and try to use Native nifi processors. ====== Executescript =============== Thread.sleep(300000) def flowFiles = session.get(100) if (!flowFiles || flowFiles.size() <= 10) { session.transfer(flowFiles, REL_SUCCESS) } else { session.transfer(flowFiles, REL_FAILURE) }     ... View more