nifi Documentation

Technical Documentation: apache/nifi

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (README.md)

Apache NiFi

Status

Resources

Contacts

Community

Features

Apache NiFi is an easy to use, powerful, and reliable system to process and distribute data.

NiFi automates cybersecurity, observability, event streams, and generative AI data pipelines and distribution
for thousands of companies worldwide across every industry.

Browser User Interface
- Seamless experience for design, control, and monitoring
- Runtime management and versioned pipelines
- Secure by default with HTTPS
Scalable Processing
- Configurable prioritization for throughput and latency
- Guaranteed delivery with retry and backoff strategies
- Horizontal scaling with clustering
Provenance Tracking
- Searchable history with configurable attributes
- Graph data lineage from source to destination
- Metadata and content for each processing decision
Extensible Design
- Plugin interface for Processors and Controller Services
- Support for Processors in native Python
- REST API for orchestration and monitoring
Secure Configuration
- Single sign-on with OpenID Connect or SAML 2
- Flexible authorization policies for role-based access
- Encrypted communication with TLS and SFTP

Requirements

NiFi supports modern operating systems and requires recent language versions for developing and running the application.

Platform Requirements

Java 21

Optional Dependencies

Python 3.10 or higher

Projects

The source repository includes several component projects.

Please review individual project documentation for additional details.

Getting Started

Project guides provide extensive documentation for installing and extending the application.

Developing

NiFi uses the Maven Wrapper for project development. The Maven Wrapper provides
shell scripts that download and cache a selected version of Apache Maven for running build
commands.

Developing on Microsoft Windows requires using mvnw.cmd instead of mvnw to run Maven commands.

Building

Run the following command to build project modules using parallel execution:

shell

./mvnw install -T1C

Run the following command to build project modules using parallel execution with static analysis to confirm compliance
with code and licensing requirements:

shell

./mvnw install -T1C -P contrib-check

Run the following command to build the application binaries without building other optional modules:

shell

./mvnw install -T1C -am -pl :nifi-assembly

Binaries

The nifi-assembly module contains the binary distribution.

shell

ls nifi-assembly/target/nifi-*-bin.zip

The nifi-assembly module includes the binary distribution in a directory for local development and testing.

shell

cd nifi-assembly/target/nifi-*-bin/nifi-*/

Running

NiFi provides shell scripts for starting and stopping the system.

Running on Microsoft Windows requires using nifi.cmd instead of nifi.sh for system commands.

Starting

Run the following command to start NiFi from the distribution directory:

shell

./bin/nifi.sh start

Accessing

The default configuration generates a random username and password on startup. NiFi writes the generated credentials
to the application log located in logs/nifi-app.log under the NiFi installation directory.

The following command can be used to find the generated credentials on operating systems with grep installed:

shell

grep Generated logs/nifi-app*log

NiFi logs the generated credentials as follows:

shell

Generated Username [USERNAME]
Generated Password [PASSWORD]

The USERNAME will be a random UUID composed of 36 characters. The PASSWORD will be a random string.

The username and password can be replaced with custom credentials using the following command:

shell

./bin/nifi.sh set-single-user-credentials <username> <password>

NiFi defaults to running on the localhost address with HTTPS on port 8443 at the following URL:

text

https://localhost:8443/nifi

Browsers will display a warning message indicating a potential security risk due to the self-signed certificate
generated during initialization. Production deployments should provision a certificate from a trusted certificate
authority and update the NiFi keystore and truststore configuration.

License

Except as otherwise noted this software is licensed under the
Apache License, Version 2.0

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Export Control

This distribution includes cryptographic software. The country in which you
currently reside may have restrictions on the import, possession, use, and/or
re-export to another country, of encryption software. BEFORE using any
encryption software, please check your country's laws, regulations and
policies concerning the import, possession, or use, and re-export of encryption
software, to see if this is permitted. See https://www.wassenaar.org for more
information.

The U.S. Government Department of Commerce, Bureau of Industry and Security
(BIS), has classified this software as Export Commodity Control Number (ECCN)
5D002.C.1, which includes information security software using or performing
cryptographic functions with asymmetric algorithms. The form and manner of this
Apache Software Foundation distribution makes it eligible for export under the
License Exception ENC Technology Software Unrestricted (TSU) exception (see the
BIS Export Administration Regulations, Section 740.13) for both object code and
source code.

The following provides more details on the included cryptographic software:

Apache NiFi uses the following libraries and frameworks for encrypted
communication and storage of sensitive information:

Nifi Extension Bundles Nifi Airtable Bundle Nifi Airtable Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Airtable.QueryAirtableTable AdditionalDetails (`nifi-extension-bundles/nifi-airtable-bundle/nifi-airtable-processors/src/main/resources/docs/org.apache.nifi.processors.airtable.QueryAirtableTable/additionalDetails.md`)

QueryAirtableTable

Description

Airtable is a spreadsheet-database hybrid. In Airtable an application is called base and each base can have multiple
tables. A table consists of records (rows) and each record can have multiple fields (columns). The QueryAirtableTable
processor can query records from a single base and table via Airtable's REST API. The processor utilizes streams to be
able to handle a large number of records. It can also split large record sets to multiple FlowFiles just like a database
processor.

Personal Access Token

Please note that API Keys were deprecated, Airtable now provides Personal Access Tokens (PATs) instead.
Airtable REST API calls requires a PAT (Personal Access Token) that needs to be passed in a request. An Airtable account
is required to generate the PAT.

API rate limit

The Airtable REST API limits the number of requests that can be sent on a per-base basis to avoid bottlenecks.
Currently, this limit is 5 requests per second per base. If this limit is exceeded you can't make another request for 30
seconds. It's your responsibility to handle this rate limit via configuring Yield Duration and Run Schedule properly. It
is recommended to start off with the default settings and to increase both parameters when rate limit issues occur.

Metadata API

Currently, the Metadata API of Airtable is unstable, and we don't provide a way to use it. Until it becomes stable you
can set up a ConvertRecord or MergeRecord processor with a JsonTreeReader to read the content and convert it into a
Record with schema.

Nifi Extension Bundles Nifi Amqp Bundle Nifi Amqp Processors Src Main Resources Docs Org.Apache.Nifi.Amqp.Processors.ConsumeAMQP AdditionalDetails (`nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/resources/docs/org.apache.nifi.amqp.processors.ConsumeAMQP/additionalDetails.md`)

ConsumeAMQP

Summary

This processor consumes messages from AMQP messaging queue and converts them to a FlowFile to be routed to the next
component in the flow. At the time of writing this document the supported AMQP protocol version is v0.9.1.

The component is based
on RabbitMQ Client API The following guide and tutorial
may also help you to brush up on some of the AMQP basics.

This processor does two things. It constructs FlwFile by extracting information from the consumed AMQP message (both
body and attributes). Once message is consumed a FlowFile is constructed. The message body is written to a FlowFile and
its com.rabbitmq.client.AMQP.BasicProperties are transfered into the FlowFile as attributes. AMQP attribute names are
prefixed with amqp$ prefix.

AMQP Properties

The following is the list of available standard AMQP properties which may come with the message: ("amqp$contentType",
"amqp$contentEncoding", "amqp$headers", "amqp$deliveryMode", "amqp$priority", "amqp$correlationId",
"amqp$replyTo", "amqp$expiration", "amqp$messageId", "amqp$timestamp", "amqp$type", "amqp$userId",
"amqp$appId", "amqp$clusterId", "amqp$routingKey")

Configuration Details

At the time of writing this document it only defines the essential configuration properties which are suitable for most
cases. Other properties will be defined later as this component progresses. Configuring PublishAMQP:

Queue - [REQUIRED] the name of AMQP queue the messages will be retrieved from. Usually provided by
administrator (e.g., 'amq.direct')
Host Name - [REQUIRED] the name of the host where AMQP broker is running. Usually provided by administrator (
e.g., 'myhost.com'). Defaults to 'localhost'.
Port - [REQUIRED] the port number where AMQP broker is running. Usually provided by the administrator (e.g., '
2453'). Defaults to '5672'.
User Name - [REQUIRED] user name to connect to AMQP broker. Usually provided by the administrator (e.g., 'me').
Defaults to 'guest'.
Password - [REQUIRED] password to use with user name to connect to AMQP broker. Usually provided by the
administrator. Defaults to 'guest'.
Use Certificate Authentication - [OPTIONAL] Use the SSL certificate common name for authentication rather than
user name/password. This can only be used in conjunction with SSL. Defaults to 'false'.
Virtual Host - [OPTIONAL] Virtual Host name which segregates AMQP system for enhanced security. Please refer
to this blog for more details on Virtual Host.

Nifi Extension Bundles Nifi Amqp Bundle Nifi Amqp Processors Src Main Resources Docs Org.Apache.Nifi.Amqp.Processors.PublishAMQP AdditionalDetails (`nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/resources/docs/org.apache.nifi.amqp.processors.PublishAMQP/additionalDetails.md`)

PublishAMQP

Summary

This processor publishes the contents of the incoming FlowFile to an AMQP-based messaging system. At the time of writing
this document the supported AMQP protocol version is v0.9.1.

The component is based
on RabbitMQ Client API The following guide and tutorial
may also help you to brush up on some of the AMQP basics.

This processor does two things. It constructs AMQP Message by extracting FlowFile contents (both body and attributes).
Once message is constructed it is sent to an AMQP Exchange. AMQP Properties will be extracted from the FlowFile and
converted to com.rabbitmq.client.AMQP.BasicProperties to be sent along with the message. Upon success the incoming
FlowFile is transferred to success Relationship and upon failure FlowFile is penalized and transferred to the
failure Relationship.

Where did my message go?

In a typical AMQP exchange model, the message that is sent to an AMQP Exchange will be routed based on the Routing Key
to its final destination in the Queue. It's called Binding. If due to some misconfiguration the binding between
the Exchange, Routing Key and the Queue is not set up, the message will have no final destination and will return (
i.e., the data will not make it to the queue). If that happens you will see a log in both app-log and bulletin stating
to that effect. Fixing the binding (normally done by AMQP administrator) will resolve the issue.

AMQP Properties

Attributes extracted from the FlowFile are considered candidates for AMQP properties if their names are prefixed with
amqp$ (e.g., amqp$contentType=text/xml). To enrich message with additional AMQP properties you may use *
UpdateAttribute* processor between the source processor and PublishAMQP processor. The following is the list of
available standard AMQP properties: ("amqp$contentType", "amqp$contentEncoding", "
amqp$headers" (if 'Headers Source' is set to 'Attribute "amqp$headers" Value') , "amqp$deliveryMode", "amqp$priority",
"amqp$correlationId", "amqp$replyTo", "amqp$expiration", "amqp$messageId", "amqp$timestamp", "amqp$type",
"amqp$userId", "amqp$appId", "amqp$clusterId")

AMQP Message Headers Source

The headers attached to AMQP message by the processor depends on the "Headers Source" property value.

Attribute "amqp$headers" Value - The processor will read single attribute "amqp$headers" and split it based on "
Header Separator" and then read headers in key=value format.
Attributes Matching Regex - The processor will pick flow file attributes by matching the regex provided in "
Attributes To Headers Regular Expression". The name of the attribute is used as key of header

Configuration Details

Exchange Name - [OPTIONAL] the name of AMQP exchange the messages will be sent to. Usually provided by the
administrator (e.g., 'amq.direct') It is an optional property. If kept empty the messages will be sent to a default
AMQP exchange. More on AMQP Exchanges could be found here.
Routing Key - [REQUIRED] the name of the routing key that will be used by AMQP to route messages from the
exchange to destination queue(s). Usually provided by administrator (e.g., 'myKey') In the event when messages are
sent to a default exchange this property corresponds to a destination queue name, otherwise a binding from the
Exchange to a Queue via Routing Key must be set (usually by the AMQP administrator). More on AMQP Exchanges and
Bindings could be found here.
Host Name - [REQUIRED] the name of the host where AMQP broker is running. Usually provided by administrator (
e.g., 'myhost.com'). Defaults to 'localhost'.
Port - [REQUIRED] the port number where AMQP broker is running. Usually provided by the administrator (e.g., '
2453'). Defaults to '5672'.
User Name - [REQUIRED] user name to connect to AMQP broker. Usually provided by the administrator (e.g., 'me').
Defaults to 'guest'.
Password - [REQUIRED] password to use with user name to connect to AMQP broker. Usually provided by the
administrator. Defaults to 'guest'.
Use Certificate Authentication - [OPTIONAL] Use the SSL certificate common name for authentication rather than
user name/password. This can only be used in conjunction with SSL. Defaults to 'false'.
Virtual Host - [OPTIONAL] Virtual Host name which segregates AMQP system for enhanced security. Please refer
to this blog for more details on Virtual Host.

Nifi Extension Bundles Nifi Asn1 Bundle Nifi Asn1 Services Src Main Resources Docs Org.Apache.Nifi.Jasn1.JASN1Reader AdditionalDetails (`nifi-extension-bundles/nifi-asn1-bundle/nifi-asn1-services/src/main/resources/docs/org.apache.nifi.jasn1.JASN1Reader/additionalDetails.md`)

JASN1Reader

Summary

This service creates record readers for ASN.1 input.

ASN.1 schema files (with full path) can be defined via the ASN.1 Files property as a comma separated list. The
controller service preprocesses these files and generates sources that it uses for parsing data later.
Note that this preprocessing may take a while, especially when the schema files are large. The service remains in
the Enabling state until this preprocessing is finished. Processors using the service are ready to be started at this
point but probably won't work properly until the service is fully Enabled.
Also note that the preprocessing phase can fail if there are problems with the ASN.1 schema files. The bulletin - as
per usual - will show error messages related to the failure but interpreting those messages may not be straightforward.
For help troubleshooting such messages please refer to the Troubleshooting section below.

The root model type can be defined via the Root Model Name property. It's format should be "MODULE-NAME.ModelType".
"MODULE-NAME" is the name of the ASN module defined at the beginnig of the ASN schema file. "ModelType" is an ASN type
defined in the ASN files that is not referenced by any other type. The reader created by this service expects ASN
records of this root model type.

More than one root model types can be defined in the ASN schema files but one service instance can only work with one
such type at a time. Multiple different ASN data types can be processed by creating multiple instances of this service.

The ASN schema files are ultimately compiled into Java classes in a temporary directory when the service is enabled.
(The directory is deleted when the service is disabled. Of course the ASN schema files remain.) The service actually
needs the fully qualified name of the class compiled from the root model type. It usually guesses the name of this class
correctly from Root Model Name.
However there may be situations where this is not the case. Should this happen, one can take use of the fact that NiFi
logs the temporary directory where the compiled Java classes can be found. Once the proper class of the root model type
is identified in that directory (should be easily done by looking for it by its name) it can be provided directly via
the Root Model Class Name property. (Note however that the service should be left Enabled while doing the search as
it deletes the temporary directory when it is disabled. To be able to set the property the service needs to be disabled
in the end - and let it remove the directory, however this shouldn't be an issue as the name of the root model class
will be the same in the new temporary directory.)

Troubleshooting

The preprocessing is done in two phases:

The first phase reads the ASN.1 schema files and parses them. Formatting errors are usually reported during this
phase. Here are some possible error messages and the potential root causes of the issues:
- line NNN:MMM: unexpected token: someFieldName - On the NNNth line, starting at the MMMth position,
  someFieldName is encountered which was unexpected. Usually this means someFieldName itself is fine but the
  previous field declaration doesn't have a comma ',' at the end.
- line NNN:MMM: unexpected token: [ - On the NNNth line, starting at the MMMth position, the opening square
  bracket '[' is encountered which was unexpected. Usually this is the index part of the field declaration and for
  some reason the field declaration is invalid. This can typically occur if the field name is invalid, e.g. starts
  with an uppercase letter. (Field names must start with a lowercase letter.)
The second phase compiles the ASN.1 schema files into Java classes. Even if the ASN.1 files meet the formal
requirements, due to the nature of the created Java files there are some extra limitations:
- On certain systems type names are treated as case-insensitive. Because of this, two types whose names only differ
  in the cases of their letters may cause errors. For example if the ASN.1 schema files define both '
  SameNameWithDifferentCase' and 'SAMENAMEWithDifferentCase', the following error may be reported:
  
  class SAMENAMEWithDifferentCase is public, should be declared in a file named SAMENAMEWithDifferentCase.java
- Certain keywords cannot be used as field names. Known reserved keywords and the corresponding reported error
  messages are:
- length
incompatible types: com.beanit.asn1bean.ber.types.BerInteger cannot be converted to
com.beanit.asn1bean.ber.BerLength
incompatible types: boolean cannot be converted to java.io.OutputStream
Some messages have been simplified; recompile with -Xdiags:verbose to get full output

Additional Preprocessing

NiFi doesn't support every feature that the ASN standard allows. To alleviate problems when encountering ASN files with
unsupported features, NiFi can do additional preprocessing steps that creates modified versions of the provided ASN
files, removing unsupported features in a way that makes them less strict but otherwise should still be compatible with
incoming data. This feature can be switched on via the 'Do Additional Preprocessing' property. The original files will
remain intact and new ones will be created with the same names in a directory set in the 'Additional Preprocessing
Output Directory' property. Please note that this is a best-effort attempt. It is also strongly recommended to compare
the resulting ASN files to the originals and make sure they are still appropriate.

The following modification are applied:

Constraints - Advanced constraints are not recognized as valid ASN elements by NiFi. This step will try to remove
all types of constraints.
E.g.

field [3] INTEGER(SIZE(1..8,...,10|12|20)) OPTIONAL

will be changed to

field [3] INTEGER OPTIONAL
Version brackets - NiFi will try to remove all version brackets and leave all defined fields as OPTIONAL.
E.g.

MyType ::= SEQUENCE {
integerField1 INTEGER,
integerField2 INTEGER,
..., -- comment1
[[ -- comment2
integerField3 INTEGER,
integerField4 INTEGER,
integerField5 INTEGER ]]
}

will be changed to

MyType ::= SEQUENCE {
integerField1 INTEGER,
integerField2 INTEGER,
..., -- comment1

-- comment2
integerField3 INTEGER OPTIONAL,
integerField4 INTEGER OPTIONAL,
integerField5 INTEGER OPTIONAL

}
"Hugging" comments - This is not really an ASN feature but a potential error. The double dash comment indicator "--"
should be separated from ASN elements.
E.g.

field [0] INTEGER(1..8)--comment

will be changed to

field [0] INTEGER(1..8) --comment

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Kinesis Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Kinesis.ConsumeKinesis AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/resources/docs/org.apache.nifi.processors.aws.kinesis.ConsumeKinesis/additionalDetails.md`)

ConsumeKinesis

IAM permissions required for ConsumeKinesis

You must add the following permissions to the IAM role or user configured in AWS Credentials Provider Service.

Cloudwatch permissions are needed only when Metrics Publishing is set to CloudWatch.

Service	Actions	Resources (ARNs)	Purpose
Amazon Kinesis Data Streams	DescribeStreamDescribeStreamSummaryRegisterStreamConsumer	Kinesis data stream from which `ConsumeKinesis` will process the data.`arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}`	Before attempting to read records, the consumer checks if the data stream exists, if it's active, and if the shards are contained in the data stream. Registers consumers to a shard.
Amazon Kinesis Data Streams	GetRecordsGetShardIteratorListShards	Kinesis data stream from which `ConsumeKinesis` will process the data.`arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}`	Reads records from a shard.
Amazon Kinesis Data Streams	SubscribeToShardDescribeStreamConsumer	Kinesis data stream from which `ConsumeKinesis` will process the data. Add this action only if you use enhanced fan-out (EFO) consumers.`arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}/consumer/*`	Subscribes to a shard for enhanced fan-out (EFO) consumers.
Amazon DynamoDB	CreateTableDescribeTableUpdateTableScanGetItemPutItemUpdateItemDeleteItem	Lease table (metadata table in DynamoDB created by `ConsumeKinesis`.`arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}`	These actions are required for `ConsumeKinesis` to manage the lease table created in DynamoDB.
Amazon DynamoDB	CreateTableDescribeTableScanGetItemPutItemUpdateItemDeleteItem	Worker metrics and coordinator state table (metadata tables in DynamoDB) created by `ConsumeKinesis`.arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-WorkerMetricStats``arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-CoordinatorState	These actions are required for `ConsumeKinesis` to manage the worker metrics and coordinator state metadata tables in DynamoDB.
Amazon DynamoDB	Query	Global secondary index on the lease table.`arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}/index/*`	This action is required for `ConsumeKinesis` to read the global secondary index of the lease table created in DynamoDB.
Amazon CloudWatch	PutMetricData	`*`	Upload metrics to CloudWatch that are useful for monitoring the application. The asterisk (*) is used because there is no specific resource in CloudWatch on which the PutMetricData action is invoked.

The following is an example policy document for ConsumeKinesis.

text

/* Detailed source-code truncated for AI context efficiency. */

Note: Replace "{Region}", "{Account}", "{Stream Name}", and "{Application Name}" in the ARNs with your own AWS region,
AWS account number, Kinesis data stream name, and ConsumeKinesis Application Name property respectively.

Consumer Type

Comparison of different Consumer Types from Amazon Kinesis Streams documentation:

Characteristics	Shared throughput consumers without enhanced fan-out	Enhanced fan-out consumers
Read throughput	Fixed at a total of 2 MB/sec per shard. If there are multiple consumers reading from the same shard, they all share this throughput. The sum of the throughputs they receive from the shard doesn't exceed 2 MB/sec.	Scales as consumers register to use enhanced fan-out. Each consumer registered to use enhanced fan-out receives its own read throughput per shard, up to 2 MB/sec, independently of other consumers.
Message propagation delay	An average of around 200 ms if you have one consumer reading from the stream. This average goes up to around 1000 ms if you have five consumers.	Typically an average of 70 ms whether you have one consumer or five consumers.
Cost	Not applicable	There is a data retrieval cost and a consumer-shard hour cost. For more information, see Amazon Kinesis Data Streams Pricing.

Record processing

When Processing Strategy property is set to RECORD, ConsumeKinesis operates in Record mode.
In this mode, the processor reads records from Kinesis streams using the configured Record Reader,
and writes them to FlowFiles using the configured Record Writer.

The processor tries to optimize the number of FlowFiles created by batching multiple records with the same schema
into a single FlowFile.

Schema changes

ConsumeKinesis supports dynamically changing Kinesis record schemas. When a record with new schema is encountered,
the currently open FlowFile is closed and a new FlowFile is created for the new schema. Thanks to this, the processor
preserves record ordering for a single Shard, even with record schema changes.

Please note, the processor relies on Record Reader to provide correct schema for each record.
Using a Reader with schema inference may produce a lot of different schemas, which may lead to excessive FlowFile creation.
If performance is a concern, it is recommended to use a Reader with a predefined schema or schema registry.

Output Strategies

This processor offers multiple strategies configured via processor property Output Strategy for converting Kinesis records into FlowFiles.

Use Content as Value (the default) writes only a Kinesis record value to a FlowFile.
Use Wrapper writes a Kinesis Record value as well as metadata into separate fields of a FlowFile record.
Inject Metadata writes a Kinesis Record value to a FlowFile record and adds a sub-record to it with metadata.

The written metadata includes the following fields:

stream: The name of the Kinesis stream the record was received from.
shardId: The identifier of the shard the record was received from.
sequenceNumber: The sequence number of the record.
subSequenceNumber: The subsequence number of the record, used when multiple smaller records are aggregated into a single Kinesis record. If a record was not part of a batch, this value will be 0.
shardedSequenceNumber: A combination of the sequence number and subsequence number. This can be used to uniquely identify a record within a shard.
partitionKey: The partition key of the record.
approximateArrival: The approximate arrival timestamp of the record (in milliseconds since epoch).

The record schema that is used when Use Wrapper is selected is as follows (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    {
      "name": "value",
      "type": [
        {
          < Fields as determined by the Record Reader for a Kinesis message >
        },
        "null"
      ]
    },
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

The record schema that is used when "Inject Metadata" is selected is as follows (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    < Fields as determined by the Record Reader for a Kinesis message >,
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy Use Wrapper is selected:

json

[
  {
    "value": {
      "address": "1234 First Street",
      "zip": "12345",
      "account": {
        "name": "Acme",
        "number": "AC1234"
      }
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]

Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy Inject Metadata is selected:

json

[
  {
    "address": "1234 First Street",
    "zip": "12345",
    "account": {
      "name": "Acme",
      "number": "AC1234"
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Parameter Providers Src Main Resources Docs Org.Apache.Nifi.Parameter.Aws.AwsSecretsManagerParameterProvider AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-parameter-providers/src/main/resources/docs/org.apache.nifi.parameter.aws.AwsSecretsManagerParameterProvider/additionalDetails.md`)

AWSSecretsManagerParameterProvider

Mapping AWS Secrets to Parameter Contexts

The AwsSecretsManagerParameterProvider maps a Secret to a Parameter Context, with key/value pairs in the Secret mapping
to parameters. To create a compatible secret from the AWS Console:

From the Secrets Manager service, click the "Store a new Secret" button
Select "Other type of secret"
Under "Key/value", enter your parameters, with the parameter names being the keys and the parameter values being the
values. Click Next.
Enter the Secret name. This will determine which Parameter Context receives the parameters. Continue through the rest
of the wizard and finally click the "Store" button.

Alternatively, from the command line, run a command like the following:

aws secretsmanager create-secret --name "[Context]" --secret-string '{ "[Param]": "[secretValue]", "[Param2]": "
[secretValue2]" }'

In this example, [Context] should be the intended name of the Parameter Context, [Param] and [Param2] should be
parameter names, and [secretValue] and [secretValue2] should be the values of each respective parameter.

Plain Text Secrets

Secrets that are not stored as JSON key/value pairs are also supported. When a secret value is not a JSON object (for
example, a PEM-encoded private key or any other plain text string), it is treated as a single parameter whose name is the
secret name and whose value is the entire secret string. The secret name is also used as the Parameter Context name.

For example, to store a PEM key as a secret:

aws secretsmanager create-secret --name "my-private-key" --secret-string file://path/to/key.pem

This produces a Parameter Context named "my-private-key" containing a single parameter also named "my-private-key" with
the contents of the PEM file as the value.

Both JSON and plain text secrets can be mixed within the same Secret Name Pattern. Each secret is automatically detected
and handled according to its format.

Configuring the Parameter Provider

AWS Secrets must be explicitly matched in the "Secret Name Pattern" property in order for them to be fetched. This
prevents more than the intended Secrets from being pulled into NiFi.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Dynamodb.PutDynamoDBRecord AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.dynamodb.PutDynamoDBRecord/additionalDetails.md`)

PutDynamoDBRecord

Description

PutDynamoDBRecord intends to provide the capability to insert multiple Items into a DynamoDB table from a
record-oriented FlowFile. Compared to the PutDynamoDB, this processor is capable to process data based other than JSON
format too and prepared to add multiple fields for a given Item. Also, PutDynamoDBRecord is designed to insert bigger
batches of data into the database.

Data types

The list data types supported by DynamoDB does not fully overlap with the capabilities of the Record data structure.
Some conversions and simplifications are necessary during inserting the data. These are:

Numeric values are stored using a floating-point data structure within Items. In some cases this representation might
cause issues with the accuracy.
Char is not a supported type within DynamoDB, these fields are converted into String values.
Enum types are stored as String fields, using the name of the given enum.
DynamoDB stores time and date related information as Strings.
Internal record structures are converted into maps.
Choice is not a supported data type, regardless of the actual wrapped data type, values enveloped in Choice are
handled as Strings.
Unknown data types are handled as stings.

Limitations

Working with DynamoDB when batch inserting comes with two inherit limitations. First, the number of inserted Items is
limited to 25 in any case. In order to overcome this, during one execution, depending on the number of records in the
incoming FlowFile, PutDynamoDBRecord might attempt multiple insert calls towards the database server. Using this
approach, the flow does not have to work with this limitation in most cases.

Having multiple external actions comes with the risk of having an unforeseen result at one of the steps. For example
when the incoming FlowFile is consists of 70 records, it will be split into 3 chunks, with a single insert operation for
every chunk. The first two chunks contains 25 Items to insert per chunk, and the third contains the remaining 20. In
some cases it might occur that the first two insert operation succeeds but the third one fails. In these cases we
consider the FlowFile "partially processed" and we will transfer it to the "failure" or "unprocessed" Relationship
according to the nature of the issue. In order to keep the information about the successfully processed chunks the
processor assigns the "dynamodb.chunks.processed" attribute to the FlowFile, which has the number of successfully
processed chunks as value.

The most common reason for this behaviour comes from the other limitation the inserts have with DynamoDB: the database
has a build in supervision over the amount of inserted data. When a client reaches the "throughput limit", the server
refuses to process the insert request until a certain amount of time. More
information here.
From the perspective of the PutDynamoDBRecord we consider these cases as temporary issues and the FlowFile will be
transferred to the "unprocessed" Relationship after which the processor will yield in order to avoid further throughput
issues. (Other kinds of failures will result transfer to the "failure" Relationship)

Retry

It is suggested to loop back the "unprocessed" Relationship to the PutDynamoDBRecord in some way. FlowFiles
transferred to that relationship considered as healthy ones might be successfully processed in a later point. It is
possible that the FlowFile contains such a high number of records, what needs more than two attempts to fully insert.
The attribute "dynamodb.chunks.processed" is "rolled" through the attempts, which means, after each trigger it will
contain the sum number of inserted chunks making it possible for the later attempts to continue from the right point
without duplicated inserts.

Partition and sort keys

The processor supports multiple strategies for assigning partition key and sort key to the inserted Items. These are:

Partition Key Strategies

Partition By Field

The processors assign one of the record fields as partition key. The name of the record field is specified by the "
Partition Key Field" property and the value will be the value of the record field with the same name.

Partition By Attribute

The processor assigns the value of a FlowFile attribute as partition key. With this strategy all the Items within a
FlowFile will share the same partition key value, and it is suggested to use for tables also having a sort key in order
to meet the primary key requirements of the DynamoDB. The property "Partition Key Field" defines the name of the Item
field and the property "Partition Key Attribute" will specify which attribute's value will be assigned to the partition
key. With this strategy the "Partition Key Field" must be different from the fields consisted by the incoming records.

Generated UUID

By using this strategy the processor will generate a UUID identifier for every single Item. This identifier will be used
as value for the partition key. The name of the field used as partition key is defined by the property "Partition Key
Field". With this strategy the "Partition Key Field" must be different from the fields consisted by the incoming
records. When using this strategy, the partition key in the DynamoDB table must have String data type.

Sort Key Strategies

None

No sort key will be assigned to the Item. In case of the table definition expects it, using this strategy will result
unsuccessful inserts.

Sort By Field

The processors assign one of the record fields as sort key. The name of the record field is specified by the "Sort Key
Field" property and the value will be the value of the record field with the same name. With this strategy the "Sort Key
Field" must be different from the fields consisted by the incoming records.

Generate Sequence

The processor assigns a generated value to every Item based on the original record's position in the incoming FlowFile (
regardless of the chunks). The first Item will have the sort key 1, the second will have sort key 2 and so on. The
generated keys are unique within a given FlowFile. The name of the record field is specified by the "Sort Key Field"
attribute. With this strategy the "Sort Key Field" must be different from the fields consisted by the incoming records.
When using this strategy, the sort key in the DynamoDB table must have Number data type.

Examples

Using fields as partition and sort key

Setup

Partition Key Strategy: Partition By Field
Partition Key Field: class
Sort Key Strategy: Sort By Field
Sort Key Field: size

Note: both fields have to exist in the incoming records!

Result

Using this pair of strategies will result Items identical to the incoming record (not counting the representational
changes from the conversion). The field specified by the properties are added to the Items normally with the only
difference of flagged as (primary) key items.

Input

json

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  }
]

Output (stylized)

type: String field with value "A"
subtype: Number field with value 4
class: String field with value "t" and serving as partition key
size: Number field with value 1 and serving as sort key

Using FlowFile filename as partition key with generated sort key

Setup

Partition Key Strategy: Partition By Attribute
Partition Key Field: source
Partition Key Attribute: filename
Sort Key Strategy: Generate Sequence
Sort Key Field: sort

Result

The FlowFile's filename attribute will be used as partition key. In this case all the records within the same FlowFile
will share the same partition key. In order to avoid collusion, if FlowFiles contain multiple records, using sort key is
suggested. In this case a generated sequence is used which is guaranteed to be unique within a given FlowFile.

Input

json

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  },
  {
    "type": "B",
    "subtype": 5,
    "class": "m",
    "size": 2
  }
]

Output (stylized)

First Item

source: String field with value "data46362.json" and serving as partition key
type: String field with value "A"
subtype: Number field with value 4
class: String field with value "t"
size: Number field with value 1
sort: Number field with value 1 and serving as sort key

Second Item

source: String field with value "data46362.json" and serving as partition key
type: String field with value "B"
subtype: Number field with value 5
class: String field with value "m"
size: Number field with value 2
sort: Number field with value 2 and serving as sort key

Using generated partition key

Setup

Partition Key Strategy: Generated UUID
Partition Key Field: identifier
Sort Key Strategy: None

Result

A generated UUID will be used as partition key. A different UUID will be generated for every Item.

Input

json

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  }
]

Output (stylized)

identifier: String field with value "872ab776-ed73-4d37-a04a-807f0297e06e" and serving as partition key
type: String field with value "A"
subtype: Number field with value 4
class: String field with value "t"
size: Number field with value 1

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Kinesis.Stream.ConsumeKinesisStream AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream/additionalDetails.md`)

ConsumeKinesisStream

Streaming Versus Batch Processing

ConsumeKinesisStream retrieves all Kinesis Records that it encounters in the configured Kinesis Stream. There are two
common, broadly defined use cases.

Per-Message Use Case

By default, the Processor will create a separate FlowFile for each Kinesis Record (message) in the Stream and add
attributes for shard id, sequence number, etc.

Per-Batch Use Case

Another common use case is the desire to process all Kinesis Records retrieved from the Stream in a batch as a single
FlowFile.

The ConsumeKinesisStream Processor can optionally be configured with a Record Reader and Record Writer. When a Record
Reader and Record Writer are configured, a single FlowFile will be created that will contain a Record for each Record
within the batch of Kinesis Records (messages), instead of a separate FlowFile per Kinesis Record.

The FlowFiles emitted in this mode will include the standard record.* attributes along with the same Kinesis Shard ID,
Sequence Number and Approximate Arrival Timestamp; but the values will relate to the last Kinesis Record that was
processed in the batch of messages constituting the content of the FlowFile.

Once a Record Writer is set the Output Strategy can be set to Use Wrapper or Use Content. When Use Wrapper is
picked the original content of the Kinesis Record will be wrapped under value key and an additional metadata
key will be populated with Stream Name, Shard ID, Partition Key, Sequence Number, Sub Sequence Number, Approximate
Arrival Timestamp and a Sharded Sequence Number that is a concatenation of Sequence and Sub Sequence Numbers that allows
a single comparison record ordering.
When Use Content is picked the original content of the Kinesis Record will be used as is.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Polly.GetAwsPollyJobStatus AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.polly.GetAwsPollyJobStatus/additionalDetails.md`)

Amazon Polly

GetAwsPollyJobStatus

Amazon Polly is a service that turns text into lifelike speech, allowing you to create applications that talk, and build
entirely new categories of speech-enabled products. Polly's Text-to-Speech (TTS) service uses advanced deep learning
technologies to synthesize natural sounding human speech. With dozens of lifelike voices across a broad set of
languages, you can build speech-enabled applications that work in many different countries.

Usage

GetAwsPollyJobStatus Processor is designed to periodically check polly job status. This processor should be used in pair
with StartAwsPollyJob Processor. If the job successfully finished it will populate outputLocation attribute of the
flow file where you can find the output of the Polly job. In case of an error failure.reason attribute will be
populated with the details.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Polly.StartAwsPollyJob AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.polly.StartAwsPollyJob/additionalDetails.md`)

Amazon Polly

StartAwsPollyJob

Usage

Amazon ML Processors are implemented to utilize ML services based on the official AWS API Reference. You can find
example json payload in the documentation at the Request Syntax sections. For more details please check the
official Polly API reference With this processor you
will trigger a startSpeechSynthesisTask async call to Polly Service. You can define json payload as property or provide
as a flow file content. Property has higher precedence. After the job is triggered the serialized json response will be
written to the output flow file. The awsTaskId attribute will be populated, so it makes it easier to query job status
by the corresponding get job status processor.

JSON payload template - note that it can be simplified with the optional fields,
check AWS documentation for more details -
example:

json

{
  "Engine": "string",
  "LanguageCode": "string",
  "LexiconNames": [
    "string"
  ],
  "OutputFormat": "string",
  "OutputS3BucketName": "string",
  "OutputS3KeyPrefix": "string",
  "SampleRate": "string",
  "SnsTopicArn": "string",
  "SpeechMarkTypes": [
    "string"
  ],
  "Text": "string",
  "TextType": "string",
  "VoiceId": "string"
}

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Textract.GetAwsTextractJobStatus AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.textract.GetAwsTextractJobStatus/additionalDetails.md`)

Amazon Textract

GetAwsTextractJobStatus

Amazon Textract is a machine learning (ML) service that automatically extracts text, handwriting, and data from scanned
documents. It goes beyond simple optical character recognition (OCR) to identify, understand, and extract data from
forms and tables.

Usage

GetAwsTextractJobStatus Processor is designed to periodically check textract job status. This processor should be used
in pair with StartAwsTextractJob Processor. FlowFile will contain the serialized Tetract response that contains the
result and additional metadata as it is documented in AWS Textract Reference.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Textract.StartAwsTextractJob AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.textract.StartAwsTextractJob/additionalDetails.md`)

Amazon Textract

StartAwsTextractJob

Usage

Amazon ML Processors are implemented to utilize ML services based on the official AWS API Reference. You can find
example json payload in the documentation at the Request Syntax sections. For more details please check the
official Textract API reference With this processor
you will trigger a startDocumentAnalysis, startDocumentTextDetection or startExpenseAnalysis async call according to
your type of textract settings. You can define json payload as property or provide as a flow file content. Property has
higher precedence. After the job is triggered the serialized json response will be written to the output flow file. The
awsTaskId attribute will be populated, so it makes it easier to query job status by the corresponding get job status
processor.

Three different type of textract task are supported: Documnet Analysis, Text Detection, Expense Analysis.

DocumentAnalysis

Starts the asynchronous analysis of an input document for relationships between detected items such as key-value pairs,
tables, and selection
elements. API Reference

Example payload:

json

{
  "ClientRequestToken": "string",
  "DocumentLocation": {
    "S3Object": {
      "Bucket": "string",
      "Name": "string",
      "Version": "string"
    }
  },
  "FeatureTypes": [
    "string"
  ],
  "JobTag": "string",
  "KMSKeyId": "string",
  "NotificationChannel": {
    "RoleArn": "string",
    "SNSTopicArn": "string"
  },
  "OutputConfig": {
    "S3Bucket": "string",
    "S3Prefix": "string"
  },
  "QueriesConfig": {
    "Queries": [
      {
        "Alias": "string",
        "Pages": [
          "string"
        ],
        "Text": "string"
      }
    ]
  }
}

ExpenseAnalysis

Starts the asynchronous analysis of invoices or receipts for data like contact information, items purchased, and vendor
names. API Reference

Example payload:

json

{
  "ClientRequestToken": "string",
  "DocumentLocation": {
    "S3Object": {
      "Bucket": "string",
      "Name": "string",
      "Version": "string"
    }
  },
  "JobTag": "string",
  "KMSKeyId": "string",
  "NotificationChannel": {
    "RoleArn": "string",
    "SNSTopicArn": "string"
  },
  "OutputConfig": {
    "S3Bucket": "string",
    "S3Prefix": "string"
  }
}

StartDocumentTextDetection

Starts the asynchronous detection of text in a document. Amazon Textract can detect lines of text and the words that
make up a line of
text. API Reference

Example payload:

json

{
  "ClientRequestToken": "string",
  "DocumentLocation": {
    "S3Object": {
      "Bucket": "string",
      "Name": "string",
      "Version": "string"
    }
  },
  "JobTag": "string",
  "KMSKeyId": "string",
  "NotificationChannel": {
    "RoleArn": "string",
    "SNSTopicArn": "string"
  },
  "OutputConfig": {
    "S3Bucket": "string",
    "S3Prefix": "string"
  }
}

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Transcribe.GetAwsTranscribeJobStatus AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.transcribe.GetAwsTranscribeJobStatus/additionalDetails.md`)

Amazon Transcribe

Automatically convert speech to text

Usage

GetAwsTranscribeJobStatus Processor is designed to periodically check Transcribe job status. This processor should be
used in pair with Transcribe Processor. FlowFile will contain the serialized Transcribe response and it will populate
the path of the output in the outputLocation attribute.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Transcribe.StartAwsTranscribeJob AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.transcribe.StartAwsTranscribeJob/additionalDetails.md`)

Amazon Transcribe

Automatically convert speech to text

Usage

Amazon ML Processors are implemented to utilize ML services based on the official AWS API Reference. You can find
example json payload in the documentation at the Request Syntax sections. For more details please check the
official Transcribe API reference With this
processor you will trigger a startTranscriptionJob async call to AWS Transcribe Service. You can define json payload as
property or provide as a flow file content. Property has higher precedence. After the job is triggered the serialized
json response will be written to the output flow file. The awsTaskId attribute will be populated, so it makes it easier
to query job status by the corresponding get job status processor.

JSON payload template - note that these can be simplified with the optional fields,
check AWS documentation for more details -
examples:

json

{
  "ContentRedaction": {
    "PiiEntityTypes": [
      "string"
    ],
    "RedactionOutput": "string",
    "RedactionType": "string"
  },
  "IdentifyLanguage": boolean,
  "IdentifyMultipleLanguages": boolean,
  "JobExecutionSettings": {
    "AllowDeferredExecution": boolean,
    "DataAccessRoleArn": "string"
  },
  "KMSEncryptionContext": {
    "string": "string"
  },
  "LanguageCode": "string",
  "LanguageIdSettings": {
    "string": {
      "LanguageModelName": "string",
      "VocabularyFilterName": "string",
      "VocabularyName": "string"
    }
  },
  "LanguageOptions": [
    "string"
  ],
  "Media": {
    "MediaFileUri": "string",
    "RedactedMediaFileUri": "string"
  },
  "MediaFormat": "string",
  "MediaSampleRateHertz": number,
  "ModelSettings": {
    "LanguageModelName": "string"
  },
  "OutputBucketName": "string",
  "OutputEncryptionKMSKeyId": "string",
  "OutputKey": "string",
  "Settings": {
    "ChannelIdentification": boolean,
    "MaxAlternatives": number,
    "MaxSpeakerLabels": number,
    "ShowAlternatives": boolean,
    "ShowSpeakerLabels": boolean,
    "VocabularyFilterMethod": "string",
    "VocabularyFilterName": "string",
    "VocabularyName": "string"
  },
  "Subtitles": {
    "Formats": [
      "string"
    ],
    "OutputStartIndex": number
  },
  "Tags": [
    {
      "Key": "string",
      "Value": "string"
    }
  ],
  "TranscriptionJobName": "string"
}

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Translate.GetAwsTranslateJobStatus AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.translate.GetAwsTranslateJobStatus/additionalDetails.md`)

Amazon Translate

GetAwsTranslateJobStatus

Amazon Translate is a neural machine translation service for translating text to and from English across a breadth of
supported languages. Powered by deep-learning technologies, Amazon Translate delivers fast, high-quality, and affordable
language translation. It provides a managed, continually trained solution, so you can easily translate company and
user-authored content or build applications that require support across multiple languages. The machine translation
engine has been trained on a wide variety of content across different domains to produce quality translations that serve
any industry need.

Usage

GetAwsTranslateJobStatus Processor is designed to periodically check translate job status. This processor should be used
in pair with Translate Processor. If the job successfully finished it will populate outputLocation attribute of the flow
file where you can find the output of the Translation job.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Ml.Translate.StartAwsTranslateJob AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.ml.translate.StartAwsTranslateJob/additionalDetails.md`)

Amazon Translate

StartAwsTranslateJob

Usage

JSON payload template - note that it can be simplified with the optional fields,
check AWS documentation for more details -
example:

json

{
  "ClientToken": "string",
  "DataAccessRoleArn": "string",
  "InputDataConfig": {
    "ContentType": "string",
    "S3Uri": "string"
  },
  "JobName": "string",
  "OutputDataConfig": {
    "EncryptionKey": {
      "Id": "string",
      "Type": "string"
    },
    "S3Uri": "string"
  },
  "ParallelDataNames": [
    "string"
  ],
  "Settings": {
    "Formality": "string",
    "Profanity": "string"
  },
  "SourceLanguageCode": "string",
  "TargetLanguageCodes": [
    "string"
  ],
  "TerminologyNames": [
    "string"
  ]
}

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.Rds.AwsRdsIamDatabasePasswordProvider AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.rds.AwsRdsIamDatabasePasswordProvider/additionalDetails.md`)

Summary

AwsRdsIamDatabasePasswordProvider generates Amazon RDS IAM authentication tokens each time a JDBC connection is requested. The Controller Service implements the NiFi DatabasePasswordProvider API, so it can be referenced from DBCP controller services to avoid storing long-lived database passwords in NiFi.

Usage

Configure an AWSCredentialsProviderControllerService so the password provider can obtain AWS credentials (for example, using an IAM role or AssumeRoleWithWebIdentity).
Create an AwsRdsIamDatabasePasswordProvider and reference the credentials provider service. Configure the AWS region. Host, port, and database user are inherited from the JDBC URL and “Database User” properties on the referencing DBCP service.
Update the DBCP controller service to set the Database Password Provider property to the new IAM provider. The static Password property is ignored when a provider is configured.
Ensure your JDBC URL enables TLS and includes the SSL parameters recommended by AWS (for example, ssl=true&sslmode=verify-full for PostgreSQL).

Each time the DBCP service needs to create a physical JDBC connection, a fresh IAM token is generated and supplied as the password. Existing pooled connections remain valid until the database closes them, so standard NiFi pooling properties such as “Maximum Connection Lifetime” still apply.

Example Setup

PostgreSQL role and privileges

Connect to the nifi database as a superuser and run:

sql

CREATE ROLE nifi_app LOGIN PASSWORD 'temporary';
GRANT rds_iam TO nifi_app;

GRANT CONNECT ON DATABASE nifi TO nifi_app;
GRANT USAGE ON SCHEMA public TO nifi_app;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO nifi_app;
GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public TO nifi_app;

ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT ALL PRIVILEGES ON TABLES TO nifi_app;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT ALL PRIVILEGES ON SEQUENCES TO nifi_app;

IAM permissions

Attach a policy like the following to your IAM role (for example myAuroraPostgresRole). Replace <region>, <account-id>, and <db-resource-id> with your values:

json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "rds-db:connect",
      "Resource": "arn:aws:rds-db:<region>:<account-id>:dbuser:<db-resource-id>/nifi_app"
    }
  ]
}

Ensure the role’s trust policy allows the NiFi host (EC2, EKS, etc.) to assume it.

CLI verification

Generate an IAM auth token and connect with psql:

bash

TOKEN=$(aws rds generate-db-auth-token \
  --hostname database-1-instance-1.ccfuwyso6lcz.us-east-1.rds.amazonaws.com \
  --port 5432 \
  --region us-east-1 \
  --username nifi_app)

PGPASSWORD="$TOKEN" psql \
  "host=database-1-instance-1.ccfuwyso6lcz.us-east-1.rds.amazonaws.com \
   port=5432 user=nifi_app dbname=nifi \
   sslmode=verify-full sslrootcert=/path/to/rds-combined-ca-bundle.pem"

When that works, configure NiFi’s DBCP service with:

Database Connection URL: jdbc:postgresql://database-1-instance-1.ccfuwyso6lcz.us-east-1.rds.amazonaws.com:5432/nifi?ssl=true&sslmode=verify-full
Database User: nifi_app
Database Password Provider: AwsRdsIamDatabasePasswordProvider

NiFi will then mint IAM tokens automatically for each new JDBC connection.

Connecting Through a Network Load Balancer in Front of an RDS Proxy

When connecting through a Network Load Balancer (NLB) placed in front of an RDS Proxy, the JDBC URL must point to the NLB endpoint, but the IAM token must be signed for the RDS Proxy endpoint. By default, the token is generated using the hostname extracted from the JDBC URL, which causes authentication to fail.

Use the optional Token Request Endpoint property to specify the RDS Proxy endpoint (hostname:port) for token signing while keeping the NLB endpoint in the JDBC URL for the actual TCP connection.

CLI verification

bash

NLB_ENDPOINT="nlb-example.elb.us-east-1.amazonaws.com"
RDS_PROXY_ENDPOINT="my-proxy.proxy-example.us-east-1.rds.amazonaws.com"

TOKEN=$(aws rds generate-db-auth-token \
  --hostname "$RDS_PROXY_ENDPOINT" \
  --port 3306 \
  --region us-east-1 \
  --username nifi_app)

mysql -h "$NLB_ENDPOINT" -P 3306 -u nifi_app --password="$TOKEN"

When that works, configure NiFi's DBCP service with:

Database Connection URL: jdbc:mysql://nlb-example.elb.us-east-1.amazonaws.com:3306/mydb
Database User: nifi_app
Database Password Provider: AwsRdsIamDatabasePasswordProvider
Token Request Endpoint: my-proxy.proxy-example.us-east-1.rds.amazonaws.com:3306

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.S3.ListS3 AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.s3.ListS3/additionalDetails.md`)

ListS3

Streaming Versus Batch Processing

ListS3 performs a listing of all S3 Objects that it encounters in the configured S3 bucket. There are two common,
broadly defined use cases.

Streaming Use Case

By default, the Processor will create a separate FlowFile for each object in the bucket and add attributes for filename,
bucket, etc. A common use case is to connect ListS3 to the FetchS3 processor. These two processors used in conjunction
with one another provide the ability to easily monitor a bucket and fetch the contents of any new object as it lands in
S3 in an efficient streaming fashion.

Batch Use Case

Another common use case is the desire to process all newly arriving objects in a given bucket, and to then perform some
action only when all objects have completed their processing. The above approach of streaming the data makes this
difficult, because NiFi is inherently a streaming platform in that there is no "job" that has a beginning and an end.
Data is simply picked up as it becomes available.

To solve this, the ListS3 Processor can optionally be configured with a Record Writer. When a Record Writer is
configured, a single FlowFile will be created that will contain a Record for each object in the bucket, instead of a
separate FlowFile per object. See the documentation for ListFile for an example of how to build a dataflow that allows
for processing all the objects before proceeding with any other step.

One important difference between the data produced by ListFile and ListS3, though, is the structure of the Records that
are emitted. The Records emitted by ListFile have a different schema than those emitted by ListS3. ListS3 emits records
that follow the following schema (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    {
      "name": "key",
      "type": "string"
    },
    {
      "name": "bucket",
      "type": "string"
    },
    {
      "name": "owner",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "etag",
      "type": "string"
    },
    {
      "name": "lastModified",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "size",
      "type": "long"
    },
    {
      "name": "storageClass",
      "type": "string"
    },
    {
      "name": "latest",
      "type": "boolean"
    },
    {
      "name": "versionId",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "tags",
      "type": [
        "null",
        {
          "type": "map",
          "values": "string"
        }
      ]
    },
    {
      "name": "userMetadata",
      "type": [
        "null",
        {
          "type": "map",
          "values": "string"
        }
      ]
    }
  ]
}

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.S3.PutS3Object AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.s3.PutS3Object/additionalDetails.md`)

PutS3Object

Multi-part Upload Details

The upload uses either the PutS3Object method or the PutS3MultipartUpload method. The PutS3Object method sends the file
in a single synchronous call, but it has a 5GB size limit. Larger files are sent using the PutS3MultipartUpload method.
This multipart process saves state after each step so that a large upload can be resumed with minimal loss if the
processor or cluster is stopped and restarted. A multipart upload consists of three steps:

Initiate upload
Upload the parts
Complete the upload

For multipart uploads, the processor saves state locally tracking the upload ID and parts uploaded, which must both be
provided to complete the upload. The AWS libraries select an endpoint URL based on the AWS region, but this can be
overridden with the 'Endpoint Override URL' property for use with other S3-compatible endpoints. The S3 API specifies
that the maximum file size for a PutS3Object upload is 5GB. It also requires that parts in a multipart upload must be at
least 5MB in size, except for the last part. These limits establish the bounds for the Multipart Upload Threshold and
Part Size properties.

Configuration Details

Object Key

The Object Key property value should not start with "/".

Credentials File

The Credentials File property allows the user to specify the path to a file containing the AWS access key and secret
key. The contents of the file should be in the following format:

text

[default]
    accessKey=<access key>
    secretKey=<security key>

Make sure the credentials file is readable by the NiFi service user.

When using the Credential File property, ensure that there are no values for the Access Key and Secret Key properties.
The Value column should read "No value set" for both. Note: Do not check "Set empty string" for either as the empty
string is considered a set value.

Nifi Extension Bundles Nifi Aws Bundle Nifi Aws Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Aws.S3.Encryption.StandardS3EncryptionService AdditionalDetails (`nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.s3.encryption.StandardS3EncryptionService/additionalDetails.md`)

S3EncryptionService

Description

The StandardS3EncryptionService manages an encryption strategy and applies that strategy to various S3 operations.
Note: This service has no effect when a processor has the Server Side Encryption property set. To use this service
with processors so configured, first create a service instance, set the Encryption Strategy to Server-side S3,
disable the Server Side Encryption processor setting, and finally, associate the processor with the service.

Configuration Details

Encryption Strategy

The name of the specific encryption strategy for this service to use when encrypting and decrypting S3 operations.

None - no encryption is configured or applied.
Server-side S3 - encryption and decryption is managed by S3; no keys are required.
Server-side KMS - encryption and decryption are performed by S3 using the configured KMS key.
Server-side Customer Key - encryption and decryption are performed by S3 using the supplied customer key.
Client-side KMS - like the Server-side KMS strategy, with the encryption and decryption performed by the client.
Client-side Customer Key - like the Server-side Customer Key strategy, with the encryption and decryption performed
by the client.

Key ID or Key Material

When configured for either the Server-side or Client-side KMS strategies, this field should contain the KMS Key ID.

When configured for either the Server-side or Client-side Customer Key strategies, this field should contain the key
material, and that material must be base64 encoded.

All other encryption strategies ignore this field.

KMS Region

KMS key region, if any. This value must match the actual region of the KMS key if supplied.

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Parameter Providers Src Main Resources Docs Org.Apache.Nifi.Parameter.Azure.AzureKeyVaultSecretsParameterProvider AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-parameter-providers/src/main/resources/docs/org.apache.nifi.parameter.azure.AzureKeyVaultSecretsParameterProvider/additionalDetails.md`)

AzureKeyVaultSecretsParameterProvider

Mapping Azure Key Vault Secrets to Parameter Contexts

The AzureKeyVaultSecretsParameterProvider maps a Secret to a Parameter, which can be grouped by adding a "group-name"
tag. To create a compatible secret from the Azure Portal:

Go to the "Key Vault" service
Create your own Key Vault
In your own Key Vault, navigate to "Secrets"
Create a secret with the name corresponding to the parameter name, and the value corresponding to the parameter
value. Under "Tags", add a tag with a Key of "group-name" and a value of the intended Parameter Group name.

Alternatively, from the command line, run a command like the following once you have a Key Vault:

text

az keyvault secret set --vault-name [Vault Name] --name [Parameter Name] --value [Parameter Value] --tags group-name=[Parameter Group Name]

In this example, [Parameter Group Name] should be the intended name of the Parameter Group, [Parameter Name] should
be the parameter name, and [Parameter Value] should be the value of the parameter. [Vault Name] should be the name
you chose for your Key Vault in Azure

Configuring the Parameter Provider

Azure Key Vault Secrets must be explicitly matched in the "Group Name Pattern" property in order for them to be fetched.
This prevents more than the intended Secrets from being pulled into NiFi.

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Azure.Storage.PutAzureDataLakeStorage AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/docs/org.apache.nifi.processors.azure.storage.PutAzureDataLakeStorage/additionalDetails.md`)

PutAzureDataLakeStorage

This processor is responsible for uploading files to Azure Data Lake Storage Gen2.

File uploading and cleanup process in case of "Write and Rename" strategy

New file upload

A temporary file is created with random prefix under the given path in '_nifitempdirectory'.
Content is appended to temp file.
Temp file is moved to the final destination directory and renamed to its original name.
In case of appending or renaming failure, the temp file is deleted.
In case of temporary file deletion failure, the temp file remains on the server.

Existing file upload

Processors with "fail" conflict resolution strategy will direct the FlowFile to "Failure" relationship.
Processors with "ignore" conflict resolution strategy will direct the FlowFile to "Success" relationship.
Processors with "replace" conflict resolution strategy:

A temporary file is created with random prefix under the given path in '_nifitempdirectory'.
Content is appended to temp file.
Temp file is moved to the final destination directory and renamed to its original name, the original file is
overwritten.
In case of appending or renaming failure, the temp file is deleted and the original file remains intact.
In case of temporary file deletion failure, both temp file and original file remain on the server.

File uploading and cleanup process in case of "Simple Write" strategy

New file upload

An empty file is created at its final destination.
Content is appended to the file.
In case of appending failure, the file is deleted.
In case of file deletion failure, the file remains on the server.

Existing file upload

Processors with "fail" conflict resolution strategy will direct the FlowFile to "Failure" relationship.
Processors with "ignore" conflict resolution strategy will direct the FlowFile to "Success" relationship.
Processors with "replace" conflict resolution strategy:

An empty file is created at its final destination, the original file is overwritten.
Content is appended to the file.
In case of appending failure, the file is deleted and the original file is not restored.
In case of file deletion failure, the file remains on the server.

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Processors Src Main Resources Docs Org.Apache.Nifi.Services.Azure.AzureEntraDatabasePasswordProvider AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/docs/org.apache.nifi.services.azure.AzureEntraDatabasePasswordProvider/additionalDetails.md`)

Summary

AzureEntraDatabasePasswordProvider acquires a short-lived Microsoft Entra access token and supplies it as the
database credential for a DBCP service. The configured Database Target selects both the Microsoft Entra scope and
how DBCP applies the credential when it creates a new physical JDBC connection.

Existing pooled connections are not reauthenticated when the token expires.

Usage

Configure an Azure Credentials Service that can obtain credentials for the public Azure cloud.
Create and enable AzureEntraDatabasePasswordProvider.
Set Azure Credentials Service to the configured credentials service.
Set Database Target to the desired JDBC credential behavior.
Configure the DBCP service with the JDBC URL, driver, database user, and Database Password Provider set to
AzureEntraDatabasePasswordProvider.
Run Verify on the provider, then run Verify on the DBCP service.

Create the Microsoft Entra principal in the database separately and grant the database privileges required by NiFi.

Workload Identity Federation

For Workload Identity Federation, configure StandardAzureCredentialsControllerService with Credentials Strategy
set to Identity Federation and select a StandardAzureIdentityFederationTokenProvider. The federation token
provider accepts an OAuth2 access token provider that supplies the external client assertion. The federated identity
credential in Microsoft Entra must match the assertion issuer, subject, and audience.

This provider does not depend on a particular external assertion issuer.

Database Targets

Database Target	Scope	DBCP placement
`AZURE_OSS_RDBMS`	`https://ossrdbms-aad.database.windows.net/.default`	Supplies the token as the JDBC password
`SQL_SERVER`	`https://database.windows.net/.default`	Supplies the token using JDBC property `accessToken`

For AZURE_OSS_RDBMS, DBCP supplies the acquired token as the JDBC password and otherwise leaves the configured user
and dynamic properties unchanged.

PostgreSQL Configuration

Configure Microsoft Entra authentication on Azure Database for PostgreSQL Flexible Server and create the database role
that corresponds to the Entra principal. Set the DBCP Database User to that mapped role name.

The token is sent using PostgreSQL cleartext-password authentication and must be protected by TLS. Use direct port 5432
as the baseline and set sslmode=verify-full, with the client configured to trust the server certificate chain.

Setting	Value
Driver Class Name	`org.postgresql.Driver`
JDBC URL	`jdbc:postgresql://<SERVER>.postgres.database.azure.com:5432/<DATABASE>?sslmode=verify-full`
Database User	Microsoft Entra principal's mapped PostgreSQL role name

Azure's built-in PgBouncer on port 6432 must be validated separately. Large Entra tokens containing many group claims
can exceed intermediary protocol buffers even when the same token works against direct port 5432. This provider does
not claim unrestricted PgBouncer compatibility.

MySQL Configuration

Configure Microsoft Entra authentication on Azure Database for MySQL Flexible Server and create an Entra user or alias
for the principal. Set the DBCP Database User to the mapped alias. MySQL user names are limited to 32 characters, so
an alias can be required for a longer Entra principal name.

MySQL Connector/J must use TLS, verify the server certificate, and permit the mysql_clear_password authentication
mechanism used to transmit the token. Configure Connector/J with a trust store containing the issuing CA certificate,
use sslMode=VERIFY_IDENTITY, and set allowCleartextPasswords=true. The Azure MySQL JDBC authentication plugin is not
required when this provider supplies the token.

Setting	Value
Driver Class Name	`com.mysql.cj.jdbc.Driver`
Driver Location(s)	compatible MySQL Connector/J driver jar provided to the DBCP service
JDBC URL	`jdbc:mysql://<SERVER>.mysql.database.azure.com:3306/<DATABASE>?sslMode=VERIFY_IDENTITY&allowCleartextPasswords=true`
Database User	Microsoft Entra principal's mapped MySQL user or alias

For SQL_SERVER, DBCP applies the token only to the physical connection attempt being created. It removes copied JDBC
properties whose names equal user, userName, or password case-insensitively, sets canonical user and
password properties to empty strings, and sets the token on JDBC property accessToken.

When SQL_SERVER is selected, DBCP preserves configured authentication and integratedSecurity properties so the
JDBC driver remains responsible for accepting or rejecting conflicting settings.

Do not configure an unprefixed dynamic property named accessToken. Commons DBCP diagnostic redaction recognizes
user and password fields, but not arbitrary access token properties. This provider supplies the token at physical
connection time so it does not need to be stored in retained DBCP configuration.

Verify and Troubleshooting

AzureEntraDatabasePasswordProvider Verify checks that the configured Azure Credentials Service can resolve
credentials and acquire a nonblank token for the selected Database Target scope. It does not validate the JDBC
URL, database server, network path, TLS configuration, driver, principal mapping, or database grants.

DBCP Verify checks the actual database connection using the configured JDBC URL, driver, database user, and any
other DBCP properties.

If provider Verify fails, confirm that the Azure Credentials Service and any upstream identity-federation services
are enabled and configured for the intended Entra application and tenant. If provider Verify succeeds but DBCP
Verify fails, investigate the JDBC connection, network, TLS, driver, database principal mapping, grants, and any
driver-specific conflicts involving authentication or integratedSecurity.

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Processors Src Main Resources Docs Org.Apache.Nifi.Services.Azure.StandardAzureIdentityFederationTokenProvider AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/docs/org.apache.nifi.services.azure.StandardAzureIdentityFederationTokenProvider/additionalDetails.md`)

StandardAzureIdentityFederationTokenProvider

The StandardAzureIdentityFederationTokenProvider provides Azure TokenCredential for workload identity federation. It exchanges tokens from external identity providers for Azure AD credentials using Azure Identity SDK's ClientAssertionCredential. This approach provides built-in token caching, automatic refresh, and robust error handling.

Components such as the ADLS and Azure Storage credentials controller services reference this provider when the Credentials Type is set to Identity Federation.

Configuration workflow

Client Assertion Provider – Select a controller service that retrieves the external workload identity token. The token is passed to Azure AD as the client_assertion parameter.
Tenant ID and Client ID – Provide the Microsoft Entra tenant and application (client) ID for the federated app registration.

At runtime the service provides a TokenCredential backed by ClientAssertionCredential. When the consuming component requests a token (specifying the appropriate scope), the credential exchanges the client assertion for an Azure AD access token via https://login.microsoftonline.com/<tenant>/oauth2/v2.0/token. The Azure Identity SDK handles token caching and automatic refresh when tokens expire.

Ensure the federated app registration has the necessary Azure RBAC roles (for example Storage Blob Data Contributor and Azure Event Hubs Data Receiver/Sender as appropriate) and that the client assertion provider refreshes assertions before they expire so new Azure access tokens can be obtained.

Azure Resource Scopes

Different Azure services require different scopes when requesting tokens. The scope is determined automatically by the consuming component based on the Azure service being accessed:

https://storage.azure.com/.default – Azure Storage operations (Blob, ADLS, Queue).
https://eventhubs.azure.net/.default – Event Hubs operations.
https://management.azure.com/.default – Azure Resource Manager APIs.

Note: Microsoft Entra requires a single resource (*.default) per client credentials request.

Event Hub components

GetAzureEventHub, PutAzureEventHub, and ConsumeAzureEventHub support the Identity Federation authentication strategy for Event Hubs connections.
ConsumeAzureEventHub also supports Identity Federation for the Blob Storage checkpoint store.

Entra ID setup summary

Create or reuse an app registration for NiFi in Microsoft Entra ID.
Add a federated credential (Certificates & secrets → Federated credentials) matching your issuer/subject. Set the audience to api://AzureADTokenExchange.
Assign RBAC roles to that app registration, such as Storage Blob Data Reader/Storage Blob Data Contributor on the storage account.
Record the Tenant ID and Client ID for configuring the controller service in NiFi.

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Processors Src Main Resources Docs Org.Apache.Nifi.Services.Azure.Storage.ADLSCredentialsControllerService AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/docs/org.apache.nifi.services.azure.storage.ADLSCredentialsControllerService/additionalDetails.md`)

ADLSCredentialsControllerService

Azure Identity Federation Token Provider

When the Credentials Type property is set to Identity Federation, configure the Azure Identity Federation Token Provider with a controller service capable of exchanging workload identity tokens for Azure AD access tokens. The provider must return an access_token issued by Microsoft Entra ID (for example using the StandardAzureIdentityFederationTokenProvider). The access token is converted to the Azure SDK representation and cached in memory until it expires.

The Azure client instances created by this service do not perform additional token refresh on their own. Ensure the configured Azure Identity Federation Token Provider automatically refreshes tokens before they expire, and that the configured scopes or audiences grant access to the target storage resources.

Security considerations of using Expression Language for sensitive properties

Allowing Expression Language for a property has the advantage of configuring the property dynamically via FlowFile
attributes or Variable Registry entries. In case of sensitive properties, it also has a drawback of exposing sensitive
information like passwords, security keys or tokens. When the value of a sensitive property comes from a FlowFile
attribute, it travels by the FlowFile in clear text form and is also saved in the provenance repository. Variable
Registry does not support the encryption of sensitive information either. Due to these, the sensitive credential data
can be exposed to unauthorized parties.

Best practices for using Expression Language for sensitive properties:

use it only if necessary
control access to the flow and to provenance repository
encrypt disks storing FlowFiles and provenance data
if the sensitive data is a temporary token (like the SAS token), use a shorter lifetime and refresh the token
periodically

Nifi Extension Bundles Nifi Azure Bundle Nifi Azure Registry Clients Src Main Resources Docs Org.Apache.Nifi.Azure.Devops.AzureDevOpsFlowRegistryClient AdditionalDetails (`nifi-extension-bundles/nifi-azure-bundle/nifi-azure-registry-clients/src/main/resources/docs/org.apache.nifi.azure.devops.AzureDevOpsFlowRegistryClient/additionalDetails.md`)

Azure DevOps Flow Registry Client – Service Principal Setup

This component stores NiFi versioned flows in Azure DevOps Git using the Azure DevOps Git REST API. It authenticates with Microsoft Entra service principals through OAuth 2.0 client credentials.

Entra ID (Microsoft Entra) – Service Principal

App registrations → New registration → name (e.g., NiFi Git DevOps Registry Client); single-tenant is fine; no redirect URI.
Record Directory (tenant) ID and Application (client) ID.
Certificates & secrets → New client secret. Note the provided Value.

Azure DevOps

Organization must use the same tenant.
Organization Settings → Users → Add users → paste the Application (client) ID; Access Level = Basic; add to the right projects/groups (e.g., Contributors). Do not send email.

NiFi Management Controller Services

StandardWebClientServiceProvider: default timeouts are usually fine; set proxy/SSL if needed. Enable.
StandardOauth2AccessTokenProvider:
- Token Endpoint (replace with value "Directory (tenant) ID"): https://login.microsoftonline.com/{tenant-id}/oauth2/v2.0/token
- Grant Type: Client Credentials
- Client ID: Application (client) ID
- Client Secret: value from the created client secret
- Scope: https://app.vssps.visualstudio.com/.default
- Configure proxy/SSL if required.
- Optional: Verify to confirm a token can be obtained
- Enable

NiFi Flow Registry Client

Type: AzureDevOpsFlowRegistryClient
Azure DevOps API URL: https://dev.azure.com
Organization, Project, Repository Name
Default Branch: must already exist (e.g., main).
Repository Path: optional subfolder; no leading or trailing /.
Authentication Strategy: Service Principal.
OAuth2 Access Token Provider: choose the configured StandardOauth2AccessTokenProvider.
Web Client Service: choose the StandardWebClientServiceProvider.
Optionally Verify to confirm read/write.

Nifi Extension Bundles Nifi Box Bundle Nifi Box Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Box.ConsumeBoxEnterpriseEvents AdditionalDetails (`nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.ConsumeBoxEnterpriseEvents/additionalDetails.md`)

ConsumeBoxEnterpriseEvents

Box Documentation

Event Types List.

Nifi Extension Bundles Nifi Box Bundle Nifi Box Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Box.FetchBoxFile AdditionalDetails (`nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.FetchBoxFile/additionalDetails.md`)

FetchBoxFile

Fetch Box files in NiFi

Find File ID
Usually FetchBoxFile is used with ListBoxFile and 'box.id' is set.

In case 'box.id' is not available, you can find the ID of the file in the following way:
- Click on the file.
- The URL in the browser will include the File ID.
  For example, if the URL were https://app.box.com/file/1012106094023?s=ldiqjwuor2vwdxeeap2rtcz66dql89h3,
  the File ID would be 1012106094023
Set File ID in 'File ID' property

Nifi Extension Bundles Nifi Box Bundle Nifi Box Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Box.ListBoxFile AdditionalDetails (`nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.ListBoxFile/additionalDetails.md`)

ListBoxFile

List Box folders in NiFi

Find Folder ID
- Navigate to the folder to be listed in Box and enter it. The URL in your browser will include the ID at the end of
  the URL. For example, if the URL were https://app.box.com/folder/191632099757, the Folder ID would be
  191632099757
Set Folder ID in 'Folder ID' property

Nifi Extension Bundles Nifi Box Bundle Nifi Box Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Box.PutBoxFile AdditionalDetails (`nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.PutBoxFile/additionalDetails.md`)

PutBoxFile

Upload files to Box from NiFi

Find Folder ID
- Navigate to the folder in Box and enter it. The URL in your browser will include the ID at the end of the URL. For
  example, if the URL were https://app.box.com/folder/191632099757, the Folder ID would be 191632099757
Set Folder ID in 'Folder ID' property

Nifi Extension Bundles Nifi Box Bundle Nifi Box Services Src Main Resources Docs Org.Apache.Nifi.Box.Controllerservices.JsonConfigBasedBoxClientService AdditionalDetails (`nifi-extension-bundles/nifi-box-bundle/nifi-box-services/src/main/resources/docs/org.apache.nifi.box.controllerservices.JsonConfigBasedBoxClientService/additionalDetails.md`)

JsonConfigBasedBoxClientService

Setting up a Box App

This processor requires a pre-configured App under the Account owning the resources being
accessed (https://app.box.com/developers/console).

The App should have the following configuration:

If you create a new App, select 'Server Authentication (with JWT)' authentication method.
If you want to use an existing App, choose one with 'OAuth 2.0 with JSON Web Tokens (Server Authentication)' as
Authentication method.
Should have a 'Client ID' and 'Client Secret'.
'App Access Level' should be 'App + Enterprise Access'.
'Application Scopes' should have 'Write all files and folders in Box' enabled.
If 'App Actor' is 'Impersonated User', 'Advanced Features' should have 'Generate user access tokens' and 'Make API calls using the as-user header' enabled.
Under 'Add and Manage Public Keys' generate a Public/Private Keypair and download the configuration JSON file (under
App Settings). The full path of this file should be set in the 'Box Config File' property.
Note that you can only download the configuration JSON with the keypair details only once, when you generate the
keypair. Also this is the only time Box will show you the private key.
If you want to download the configuration JSON file later (under 'App Settings') - or if you want to use your own
keypair - after you download it you need to edit the file and add the keypair details manually.
After all settings are done, the App needs to be reauthorized on the admin page. ('Reauthorize App'
at https://app.box.com/master/custom-apps.)
If the app is configured for the first time it needs to be added ('Add App') before it can be authorized.

Nifi Extension Bundles Nifi Dropbox Bundle Nifi Dropbox Services Src Main Resources Docs Org.Apache.Nifi.Services.Dropbox.StandardDropboxCredentialService AdditionalDetails (`nifi-extension-bundles/nifi-dropbox-bundle/nifi-dropbox-services/src/main/resources/docs/org.apache.nifi.services.dropbox.StandardDropboxCredentialService/additionalDetails.md`)

StandardDropboxCredentialService

Generating credentials for Dropbox authentication

StandardDropboxCredentialService requires "App Key", "App Secret", "Access Token" and "Refresh Token".

This document describes how to generate these credentials using an existing Dropbox account.

Generate App Key and App Secret

Login with your Dropbox account.
If you already have an app created, go to Dropbox Developers page, click on "App
Console" button and select your app. On the app's info page you will find the "App key" and "App secret".
(See also Dropbox Getting Started, App Console tab, "
Navigating the App Console" chapter)
If you don't have any apps, go to Dropbox Developers page and click on "Create
app" button. (See also Dropbox Getting Started, App
Console tab, "Creating a Dropbox app" chapter.)
- On the next page select "Scoped access" and "Full Dropbox" as access type.
- Provide a name for your app.
- On the app's info page you will find the "App key" and "App secret". (See
  also Dropbox Getting Started, App Console tab, "
  Navigating the App Console" chapter.)

Set required permissions for your app

The "files.content.read" permission has to be enabled for the application to be able to read the files in Dropbox.

You can set permissions in Dropbox Developers page.

Click on "App Console" button and select your app.
Go to "Permissions" tab and enable the "files.content.read" permission.
Click "Submit" button.
NOTE: In case you already have an Access Token and Refresh Token, those tokens have to be regenerated after the
permission change. See "Generate Access Token and Refresh Token" chapter about token generation.

Generate Access Token and Refresh Token

Go to the following web page:

https://www.dropbox.com/oauth2/authorize?token\_access\_type=offline&response\_type=code&client\_id=_your\_app\_key_
Click "Next" and click on "Allow" button on the next page.
An access code will be generated for you, it will be displayed on the next page:

"Access Code Generated
Enter this code into your_app_name to finish the process
your_generated_access_code"
Execute the following command from terminal to fetch the access and refresh tokens.

Make sure you execute the curl command right after the access code generation, since the code expires very quickly.
In case the curl command returns "invalid grant" error, please generate a new access code (see previous step)

curl https://api.dropbox.com/oauth2/token -d code=your_generated_access_code -d grant_type=authorization_code
-u your_app_key:your_app_secret
The curl command results a json file which contains the "access_token" and "refresh_token":

json

{
  "access_token": "sl.xxxxxxxxxxx"
  "expires_in": 14400,
  "refresh_token": "xxxxxx",
  "scope": "files.content.read files.metadata.read",
  "uid": "xxxxxx",
  "account_id": "dbid:xxxx"
}

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Client Service Src Main Resources Docs Org.Apache.Nifi.Elasticsearch.ElasticSearchClientServiceImpl AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl/additionalDetails.md`)

ElasticSearchClientServiceImpl

Sniffing

The Elasticsearch Sniffer can be used to locate Elasticsearch Nodes within a Cluster to which you are connecting. This
can be beneficial if your cluster dynamically changes over time, e.g. new Nodes are added to maintain performance during
heavy load.

Sniffing can also be used to update the list of Hosts within the Cluster if a connection Failure is encountered during
operation. In order to "Sniff on Failure", you must also enable "Sniff Cluster Nodes".

Not all situations make sense to use Sniffing, for example if:

Elasticsearch is situated behind a load balancer, which dynamically routes connections from NiFi
Elasticsearch is on a different network to NiFi

There may also be need to set some of
the Elasticsearch Networking Advanced Settings,
such as network.publish_host to ensure that the HTTP Hosts found by the Sniffer are accessible by NiFi. For example,
Elasticsearch may use a network internal publish_host that is inaccessible to NiFi, but instead should use an
address/IP that NiFi understands. It may also be necessary to add this same address to Elasticsearch's
network.bind_host list.

See Elasticsearch sniffing best practices: What, when, why, how
for more details of the best practices.

Resources Usage Consideration

This Elasticsearch client relies on a RestClient using the Apache HTTP Async Client. By default, it will start one
dispatcher thread, and a number of worker threads used by the connection manager. There will be as many worker thread as
the number of locally detected processors/cores on the NiFi host. Consequently, it is highly recommended to have only
one instance of this controller service per remote Elasticsearch destination and have this controller service shared
across all the Elasticsearch processors of the NiFi flows. Having a very high number of instances could lead to resource
starvation and result in OOM errors.

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Client Service Src Main Resources Docs Org.Apache.Nifi.Elasticsearch.ElasticSearchLookupService AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchLookupService/additionalDetails.md`)

ElasticSearchLookupService

This lookup service uses ElasticSearch as its data source. Mappings in LookupRecord map record paths to paths within an
ElasticSearch document. Example:

text

/user/name => user.contact.name

That would map the record path /user/name to an embedded document named contact with a field named name.

The query that is assembled from these is a boolean query where all the criteria are under the must list. In addition,
wildcards are not supported right now and all criteria are translated into literal match queries.

Post-Processing

Because an ElasticSearch result might be structured differently than the record which will be enriched by this service,
users can specify an additional set of mappings on this lookup service that map JsonPath operations to record paths.
Example:

text

$.user.contact.email => /user/email_address

Would copy the field email from the embedded document contact into the record at that path.

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.ConsumeElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.ConsumeElasticsearch/additionalDetails.md`)

SearchElasticsearch

This processor is intended for use with the Elasticsearch JSON DSL and Elasticsearch 5.X and newer. It is designed to be
able to create a JSON query using input properties and execute it against an Elasticsearch cluster in a paginated
manner. Like all processors in the "restapi" bundle, it uses the official Elastic client APIs, so it supports leader
detection.

Results will be sorted on the field that is to be tracked, with the sort order set as a property.

Search results and aggregation results can be split up into multiple flowfiles. Aggregation results will only be split
at the top level because nested aggregations lose their context (and thus lose their value) if separated from their
parent aggregation. Additionally, the results from all pages can be combined into a single flowfile (but the processor
will only load each page of data into memory at any one time).

The following is an example query that would be created for tracking a "@timestamp" field:

json

{
  "query": {
    "size": 10000,
    "sort": {
      "@timestamp": "desc"
    },
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gt": "2023-09-01"
            }
          }
        }
      ]
    }
  }
}

Additional "filter" entries can be added as a JSON string in the query filter property, for example:

json

[
  {
    "term": {
      "department": "accounts"
    }
  },
  {
    "term": {
      "title.keyword": "Annual Report"
    }
  }
]

Query Pagination Across Processor Executions

This processor runs on a schedule in order to execute the same query repeatedly. Once a paginated query has been
initiated within Elasticsearch, this processor will continue to retrieve results for that same query until no further
results are available. After that point, a new paginated query will be initiated using the same Query JSON, but with
the "range" filter query's "gt" field set to the last value obtained from previous results.

If the results are "Combined" from this processor, then the paginated query will run continually within a single
invocation until no more results are available (then the processor will start a new paginated query upon its next
invocation). If the results are "Split" or "Per Page", then each invocation of this processor will retrieve the next
page of results until either there are no more results or the paginated query expires within Elasticsearch.

Resetting Queries / Clearing Processor State

Cluster State is used to track the progress of a paginated query within this processor. If there is need to restart the
query completely or change the processor configuration after a paginated query has already been started, be sure to
"Clear State" of the processor once it has been stopped and before restarting.

Note that clearing the processor's state will lose details of where the processor was up to with tracking documents
retrieved. Update the "Initial Value" with the appropriate value before restarting the processor to continue with where
it was up to.

Duplicate Results

This processor does not attempt to de-duplicate results between queries, for example if the same query runs twice and
(some or all of) the results are identical, the output will contain these same results for both invocations. This might
happen if the NiFi Primary Node changes while a page of data is being retrieved, or if the processor state is cleared,
then the processor is restarted.

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.DeleteByQueryElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.DeleteByQueryElasticsearch/additionalDetails.md`)

DeleteByQueryElasticsearch

This processor executes a delete operation against one or more indices using the _delete_by_query handler. The query
should be a valid Elasticsearch JSON DSL query (Lucene syntax is not supported). An example query:

json

{
  "query": {
    "match": {
      "username.keyword": "john.smith"
    }
  }
}

To delete all the contents of an index, this could be used:

json

{
  "query": {
    "match_all": {}
  }
}

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.JsonQueryElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.JsonQueryElasticsearch/additionalDetails.md`)

JsonQueryElasticsearch

This processor is intended for use with the Elasticsearch JSON DSL and Elasticsearch 5.X and newer. It is designed to be
able to take a JSON query (e.g. from Kibana) and execute it as-is against an Elasticsearch cluster. Like all processors
in the "restapi" bundle, it uses the official Elastic client APIs, so it supports leader detection.

The query JSON to execute can be provided either in the Query configuration property or in the content of the flowfile.
If the Query Attribute property is configured, the executed query JSON will be placed in the attribute provided by this
property.

Additionally, search results and aggregation results can be split up into multiple flowfiles. Aggregation results will
only be split at the top level because nested aggregations lose their context (and thus lose their value) if separated
from their parent aggregation. The following is an example query that would be accepted:

json

{
  "query": {
    "match": {
      "restaurant.keyword": "Local Pizzaz FTW Inc"
    }
  },
  "aggs": {
    "weekly_sales": {
      "date_histogram": {
        "field": "date",
        "interval": "week"
      },
      "aggs": {
        "items": {
          "terms": {
            "field": "product",
            "size": 10
          }
        }
      }
    }
  }
}

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.PaginatedJsonQueryElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PaginatedJsonQueryElasticsearch/additionalDetails.md`)

PaginatedJsonQueryElasticsearch

This processor is intended for use with the Elasticsearch JSON DSL and Elasticsearch 5.X and newer. It is designed to be
able to take a JSON query (e.g. from Kibana) and execute it as-is against an Elasticsearch cluster in a paginated
manner. Like all processors in the "restapi" bundle, it uses the official Elastic client APIs, so it supports leader
detection.

The query is paginated in Elasticsearch using one of the available methods - "Scroll" or "Search After" (optionally with
a "Point in Time" for Elasticsearch 7.10+ with XPack enabled). The number of results per page can be controlled using
the size parameter in the Query JSON. For Search After functionality, a sort parameter must be present within
the Query JSON.

The following is an example query that would be accepted:

json

{
  "query": {
    "size": 10000,
    "sort": {
      "product": "desc"
    },
    "match": {
      "restaurant.keyword": "Local Pizzaz FTW Inc"
    }
  },
  "aggs": {
    "weekly_sales": {
      "date_histogram": {
        "field": "date",
        "interval": "week"
      },
      "aggs": {
        "items": {
          "terms": {
            "field": "product",
            "size": 10
          }
        }
      }
    }
  }
}

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.PutElasticsearchJson AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md`)

PutElasticsearchJson

This processor is for accessing the Elasticsearch Bulk API. It provides the ability to configure bulk operations on a
per-FlowFile basis, which is what separates it from PutElasticsearchRecord.

As part of the Elasticsearch REST API bundle, it uses a controller service to manage connection information and that
controller service is built on top of the official Elasticsearch client APIs. That provides features such as automatic
master detection against the cluster which is missing in the other bundles.

This processor builds one Elasticsearch Bulk API body per (batch of) FlowFiles. Care should be taken to batch FlowFiles
into appropriately-sized chunks so that NiFi does not run out of memory and the requests sent to Elasticsearch are not
too large for it to handle. When failures do occur, this processor is capable of attempting to route the FlowFiles that
failed to an errors queue so that only failed FlowFiles can be processed downstream or replayed.

The index, operation and (optional) type fields are configured with default values. The ID (optional unless the
operation is "index") can be set as an attribute on the FlowFile(s).

Field Path Mode

The Identifier Field, Index Field and Timestamp Field properties each name a field within the document
whose value is extracted (and used as the document ID, index name, or @timestamp respectively). The Field Path
Mode property controls how those values are interpreted:

Literal Field Name (the default) — each value is the exact name of a top-level field. A / or \ is just
a character in the field name, so a value like @metadata/id matches a top-level field literally named
@metadata/id.
Nested Field Path — each value is a /-delimited path into nested objects, as described below.

Literal Field Name (default)

For the document:

json

{
  "@metadata/id": "abc",
  "message": "Hello, world"
}

an Identifier Field of @metadata/id selects abc — the top-level field whose name is literally
@metadata/id. The same value against {"@metadata": {"id": "abc"}} selects nothing, because there is no
top-level field with that exact name.

Nested Field Path

The remainder of this section applies when Field Path Mode is set to Nested Field Path. For the document:

json

{
  "@metadata": {
    "id": "abc",
    "index": "my-index"
  },
  "message": "Hello, world"
}

the Identifier Field path @metadata/id selects abc
the Index Field path @metadata/index selects my-index

Field names that contain a slash

Because / separates path segments, a field whose name literally contains a / must have that slash escaped as \/,
and a literal backslash must be escaped as \\. For example, for the document {"a/b": "abc"} the path a\/b selects
the top-level field named a/b.

Removal and pruning

When the corresponding Retain property is set to false, the field is removed from the document body after its
value is extracted. For a nested path, any parent object that is left empty by the removal is also pruned. For example,
extracting and removing @metadata/id from {"@metadata": {"id": "abc"}, "message": "Hello, world"} leaves
{"message": "Hello, world"}, with the now-empty @metadata object removed.

Ingest Pipeline

Documents can be routed through an Elasticsearch ingest pipeline
at index time. This applies to Index and Create operations only. The pipeline is added to the bulk action
header for each document ({"index": {"_index": "...", "pipeline": "..."}}) and can be set two ways:

Pipeline — a static pipeline name applied to every document (supports Expression Language).
Pipeline Field — the name of a field within each document whose value is the pipeline, resolved per document.
Like the Index Field, this honors the Field Path Mode property, so it can be read from a top-level field or a
nested /-delimited path. When Pipeline Field is blank or absent from a document, the static Pipeline property
is used as the fallback.

Retain Pipeline Field controls whether the field is left in the document body or removed before indexing (with
empty parent objects pruned, as described above).

For example, with Pipeline Field set to @metadata/pipeline (in Nested Field Path mode) and Retain Pipeline Field
set to false, the document:

json

{
  "@metadata": {
    "pipeline": "my-ingest-pipeline"
  },
  "message": "Hello, world"
}

is indexed as {"message": "Hello, world"} with the bulk action header "pipeline": "my-ingest-pipeline", so
different documents in the same FlowFile can be routed through different ingest pipelines based on their own content.

Dynamic Templates

Index and Create operations can use Dynamic Templates. The Dynamic Templates property must be parsable as a JSON object.

Example - Index with Dynamic Templates

json

{
  "message": "Hello, world"
}

The Dynamic Templates property below would be parsable:

json

{
  "message": "keyword_lower"
}

Would create Elasticsearch action:

json

{
  "index": {
    "_id": "1",
    "_index": "test",
    "dynamic_templates": {
      "message": "keyword_lower"
    }
  }
}

json

{
  "doc": {
    "message": "Hello, world"
  }
}

Update/Upsert Scripts

Update and Upsert operations can use a script. Scripts must contain all the elements required by Elasticsearch, e.g.
source and lang. The Script property must be parsable as a JSON object.

If a script is defined for an upset, the Flowfile content will be used as the upsert fields in the Elasticsearch action.
If no script is defined, the FlowFile content will be used as the update doc (or doc_as_upsert for upsert operations).

Example - Update without Script

json

{
  "message": "Hello, world",
  "from": "john.smith"
}

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

json

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

Example - Upsert with Script

json

{
  "counter": 1
}

The script property below would be parsable:

json

{
  "source": "ctx._source.counter += params.param1",
  "lang": "painless",
  "params": {
    "param1": 1
  }
}

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

json

{
  "script": {
    "source": "ctx._source.counter += params.param1",
    "lang": "painless",
    "params": {
      "param1": 1
    }
  },
  "upsert": {
    "counter": 1
  }
}

Bulk Action Header Fields

Dynamic Properties can be defined on the processor with BULK: prefixes. Users must ensure that only known Bulk action
fields are sent to Elasticsearch for the relevant index operation defined for the FlowFile, Elasticsearch will reject
invalid combinations of index operation and Bulk action fields.

Example - Update with Retry on Conflict

json

{
  "message": "Hello, world",
  "from": "john.smith"
}

With the Dynamic Property below:

BULK:retry_on_conflict = 3

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test",
    "retry_on_conflict": "3"
  }
}

json

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

Index Operations

Valid values for "operation" are:

create
delete
index
update
upsert

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.PutElasticsearchRecord AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchRecord/additionalDetails.md`)

PutElasticsearchRecord

This processor is for accessing the Elasticsearch Bulk API. It provides the ability to configure bulk operations on a
per-record basis which is what separates it from PutElasticsearchJson. For example, it is possible to define multiple
commands to index documents, followed by deletes, creates and update operations against the same index or other indices
as desired.

This processor builds one Elasticsearch Bulk API body per record set. Care should be taken to split up record sets into
appropriately-sized chunks so that NiFi does not run out of memory and the requests sent to Elasticsearch are not too
large for it to handle. When failures do occur, this processor is capable of attempting to write the records that failed
to an output record writer so that only failed records can be processed downstream or replayed.

Per-Record Actions

The index, operation and (optional) type fields are configured with default values that can be overridden using record
path operations that find an index or type value in the record set. The ID and operation type (create, index, update,
upsert or delete) can also be extracted in a similar fashion from the record set. A "@timestamp" field can be added to
the data either using a default or by extracting it from the record set. This is useful if the documents are being
indexed into an Elasticsearch Data Stream.

Example - per-record actions

The following is an example of a document exercising all of these features:

json

{
  "metadata": {
    "id": "12345",
    "index": "test",
    "type": "message",
    "operation": "index"
  },
  "message": "Hello, world",
  "from": "john.smith",
  "ts": "2021-12-03'T'14:00:00.000Z"
}

json

{
  "metadata": {
    "id": "12345",
    "index": "test",
    "type": "message",
    "operation": "delete"
  }
}

The record path operations below would extract the relevant data:

/metadata/id
/metadata/index
metadata/type
metadata/operation
/ts

Dynamic Templates

Index and Create operations can use Dynamic Templates from the Record, record path operations can be configured to find
the Dynamic Templates from the record set. Dynamic Templates fields in Records must either be a Map, child Record or a
string that can be parsable as a JSON object.

Example - Index with Dynamic Templates

json

{
  "message": "Hello, world",
  "dynamic_templates": "{\"message\": \"keyword_lower\"}"
}

The record path operation below would extract the relevant Dynamic Templates:

/dynamic_templates

Would create Elasticsearch action:

json

{
  "index": {
    "_id": "1",
    "_index": "test",
    "dynamic_templates": {
      "message": "keyword_lower"
    }
  }
}

json

{
  "doc": {
    "message": "Hello, world"
  }
}

Update/Upsert Scripts

Update and Upsert operations can use a script from the Record, record path operations can be configured to find the
script from the record set. Scripts must contain all the elements required by Elasticsearch, e.g. source and lang.
Script fields in Records must either be a Map, child Record or a string that can be parsable as a JSON object.

If a script is defined for an upset, any fields remaining in the Record will be used as the upsert fields in the
Elasticsearch action. If no script is defined, all Record fields will be used as the update doc (or doc_as_upsert for
upsert operations).

Example - Update without Script

json

{
  "message": "Hello, world",
  "from": "john.smith"
}

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

json

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

Example - Upsert with Script

json

{
  "counter": 1,
  "script": {
    "source": "ctx._source.counter += params.param1",
    "lang": "painless",
    "params": {
      "param1": 1
    }
  }
}

The record path operation below would extract the relevant script:

/script

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

json

{
  "script": {
    "source": "ctx._source.counter += params.param1",
    "lang": "painless",
    "params": {
      "param1": 1
    }
  },
  "upsert": {
    "counter": 1
  }
}

Bulk Action Header Fields

Dynamic Properties can be defined on the processor with BULK: prefixes. The value of the Dynamic Property is a record
path operation to find the field value from the record set. Users must ensure that only known Bulk action fields are
sent to Elasticsearch for the relevant index operation defined for the Record, Elasticsearch will reject invalid
combinations of index operation and Bulk action fields.

Example - Update with Retry on Conflict

json

{
  "message": "Hello, world",
  "from": "john.smith",
  "retry": 3
}

The Dynamic Property and record path operation below would extract the relevant field:

BULK:retry_on_conflict = /retry

Would create Elasticsearch action:

json

{
  "update": {
    "_id": "1",
    "_index": "test",
    "retry_on_conflict": 3
  }
}

json

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

Index Operations

Valid values for "operation" are:

create
delete
index
update
upsert

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.SearchElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.SearchElasticsearch/additionalDetails.md`)

SearchElasticsearch

This processor is intended for use with the Elasticsearch JSON DSL and Elasticsearch 5.X and newer. It is designed to be
able to take a JSON query (e.g. from Kibana) and execute it as-is against an Elasticsearch cluster in a paginated
manner. Like all processors in the "restapi" bundle, it uses the official Elastic client APIs, so it supports leader
detection.

The query to execute must be provided in the Query configuration property.

The following is an example query that would be accepted:

json

{
  "query": {
    "size": 10000,
    "sort": {
      "product": "desc"
    },
    "match": {
      "restaurant.keyword": "Local Pizzaz FTW Inc"
    }
  },
  "aggs": {
    "weekly_sales": {
      "date_histogram": {
        "field": "date",
        "interval": "week"
      },
      "aggs": {
        "items": {
          "terms": {
            "field": "product",
            "size": 10
          }
        }
      }
    }
  }
}

Query Pagination Across Processor Executions

Resetting Queries / Clearing Processor State

Local State is used to track the progress of a paginated query within this processor. If there is need to restart the
query completely or change the processor configuration after a paginated query has already been started, be sure to
"Clear State" of the processor once it has been stopped and before restarting.

Duplicate Results

This processor will continually run the same query unless the processor properties are updated, so unless the data in
Elasticsearch has changed, the same data will be retrieved multiple times.

Nifi Extension Bundles Nifi Elasticsearch Bundle Nifi Elasticsearch Restapi Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Elasticsearch.UpdateByQueryElasticsearch AdditionalDetails (`nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.UpdateByQueryElasticsearch/additionalDetails.md`)

UpdateByQueryElasticsearch

This processor executes an update operation against one or more indices using the _update_by_query handler. The query
should be a valid Elasticsearch JSON DSL query (Lucene syntax is not supported). An optional Elasticsearch script can be
specified to execute against the matched documents. An example query with script:

json

{
  "script": {
    "source": "ctx._source.count++",
    "lang": "painless"
  },
  "query": {
    "match": {
      "username.keyword": "john.smith"
    }
  }
}

To update all the contents of an index, this could be used:

json

{
  "query": {
    "match_all": {}
  }
}

Nifi Extension Bundles Nifi Email Bundle Nifi Email Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Email.ConsumeIMAP AdditionalDetails (`nifi-extension-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.md`)

ConsumeIMAP

This Processor consumes email messages via IMAP protocol and sends the content of an email message as content of the
FlowFile. Content of the incoming email message is written as raw bytes to the content of the outgoing FlowFile.

Different email providers may require additional Java Mail properties which could be provided as dynamic properties. For
example, below is a sample configuration for GMail:

Processor's static properties:

Host Name - imap.gmail.com
Port - 993
User Name - [your user name]
Password - [your password]
Folder - INBOX

Processor's dynamic properties:

mail.imap.socketFactory.class - javax.net.ssl.SSLSocketFactory
mail.imap.socketFactory.fallback - false
mail.store.protocol - imaps

Another useful property is mail.debug which allows Java Mail API to print protocol messages to the console helping
you to both understand what's going on and debug issues.

For the full list of available Java Mail properties please refer
to here

Nifi Extension Bundles Nifi Email Bundle Nifi Email Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Email.ConsumePOP3 AdditionalDetails (`nifi-extension-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumePOP3/additionalDetails.md`)

ConsumePOP3

This Processor consumes email messages via POP3 protocol and sends the content of an email message as content of the
FlowFile. Content of the incoming email message is written as raw bytes to the content of the outgoing FlowFile.

Since different serves may require different Java Mail properties such properties could be provided via dynamic
properties. For example, below is a sample configuration for GMail:

Processor's static properties:

Host Name - pop.gmail.com
Port - 995
User Name - [your user name]
Password - [your password]
Folder - INBOX

Processor's dynamic properties:

mail.pop3.socketFactory.class - javax.net.ssl.SSLSocketFactory
mail.pop3.socketFactory.fallback - false

Another useful property is mail.debug which allows Java Mail API to print protocol messages to the console helping
you to both understand what's going on and debug issues.

For the full list of available Java Mail properties please refer
to here

Nifi Extension Bundles Nifi Evtx Bundle Nifi Evtx Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Evtx.ParseEvtx AdditionalDetails (`nifi-extension-bundles/nifi-evtx-bundle/nifi-evtx-processors/src/main/resources/docs/org.apache.nifi.processors.evtx.ParseEvtx/additionalDetails.md`)

ParseEvtx

This processor is used to parse Windows event logs in the binary evtx format. The input flow files' content should be
evtx files. The processor has 4 outputs:

The original unmodified FlowFile
The XML resulting from parsing at the configured granularity
Failed parsing with partial output
Malformed chunk in binary form

Output XML Example:

text

/* Detailed source-code truncated for AI context efficiency. */

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Parameter Providers Src Main Resources Docs Org.Apache.Nifi.Parameter.Gcp.GcpSecretManagerParameterProvider AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-parameter-providers/src/main/resources/docs/org.apache.nifi.parameter.gcp.GcpSecretManagerParameterProvider/additionalDetails.md`)

GcpSecretManagerParameterProvider

Mapping GCP Secrets to Parameter Contexts

The GcpSecretManagerParameterProvider maps a Secret to a Parameter, which can be grouped by adding a "group-name" label.
To create a compatible secret from the GCP Console:

From the Secret Manager service, click the "Create Secret" button
Enter the Secret name. This is the name of a parameter. Enter a value.
Under "Labels", add a label with a Key of "group-name" and a value of the intended Parameter Group name.

Alternatively, from the command line, run a command like the following:

text

printf "[Parameter Value]" | gcloud secrets create --labels=group-name="[Parameter Group Name]" "[Parameter Name]" --data-file=-

In this example, [Parameter Group Name] should be the intended name of the Parameter Group, [Parameter Name] should
be the parameter name, and [Parameter Value] should be the value of the parameter.

Configuring the Parameter Provider

GCP Secrets must be explicitly matched in the "Group Name Pattern" property in order for them to be fetched. This
prevents more than the intended Secrets from being pulled into NiFi.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Bigquery.PutBigQuery AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.bigquery.PutBigQuery/additionalDetails.md`)

PutBigQuery

Streaming Versus Batching Data

PutBigQuery is record based and is relying on the gRPC based Write API using protocol buffers. The underlying stream
supports both streaming and batching approaches.

Streaming

With streaming the appended data to the stream is instantly available in BigQuery for reading. It is configurable how
many records (rows) should be appended at once. Only one stream is established per flow file so at the conclusion of the
FlowFile processing the used stream is closed and a new one is opened for the next FlowFile. Supports exactly once
delivery semantics via stream offsets.

Batching

Similarly to the streaming approach one stream is opened for each FlowFile and records are appended to the stream.
However data is not available in BigQuery until it is committed by the processor at the end of the FlowFile processing.

Improvement opportunities

The table has to exist on BigQuery side it is not created automatically
The Write API supports multiple streams for parallel execution and transactionality across streams. This is not
utilized at the moment as this would be covered on NiFI framework level.

The official Google Write API documentation provides additional
details.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Cloudsql.GcpCloudSqlIamDatabasePasswordProvider AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider/additionalDetails.md`)

Summary

GcpCloudSqlIamDatabasePasswordProvider generates a short-lived Cloud SQL IAM login token and supplies it as the
database password for a DBCP service. Use it when you want NiFi to connect to Cloud SQL without storing a long-lived
database password in NiFi.

The provider works with Cloud SQL for PostgreSQL and Cloud SQL for MySQL. When a DBCP service references this provider,
the static DBCP password property is ignored.

Usage

Configure GCPCredentialsControllerService so NiFi can obtain Google credentials.
Create and enable GcpCloudSqlIamDatabasePasswordProvider.
Set GCP Credentials Provider Service to the credentials service.
Configure the DBCP service with the JDBC URL, driver, database user, and Database Password Provider set to
GcpCloudSqlIamDatabasePasswordProvider.
Run Verify on the provider, then run Verify on the DBCP service.

Create the Cloud SQL IAM database user separately and grant the database privileges required by your application.

Workload Identity Federation

For the Snowflake Workload Identity Federation configuration tested with this provider, configure
GCPCredentialsControllerService with Target Service Account. The workload identity principal must also have
roles/iam.workloadIdentityUser on that target service account. Other Google credential configurations can be used
when they can acquire a token accepted by Cloud SQL for the configured database user.

PostgreSQL Configuration

Cloud SQL for PostgreSQL expects the DBCP Database User to match the IAM identity used for login. Configure the
DBCP service with a PostgreSQL JDBC driver and PostgreSQL JDBC URL. Configure TLS in the DBCP service as required for
your environment.

Setting	Value
Driver Class Name	`org.postgresql.Driver`
JDBC URL	`jdbc:postgresql://<HOST>:5432/<DATABASE>?sslmode=require`
Database User for Google user	full email address
Database User for service account	service-account email without `.gserviceaccount.com`

Example service-account mapping:

Target service account: [email protected]
DBCP Database User: [email protected]

MySQL Configuration

Cloud SQL for MySQL uses the full service-account email when the IAM database user is created, but the JDBC login name
must be only the portion before @. Configure the DBCP service with a compatible MySQL Connector/J driver and MySQL
JDBC URL. Configure TLS in the DBCP service as required for your environment.

Setting	Value
Driver Class Name	`com.mysql.cj.jdbc.Driver`
Driver Location(s)	compatible MySQL Connector/J driver jar provided to the DBCP service
JDBC URL	`jdbc:mysql://<HOST>:3306/<DATABASE>?sslMode=REQUIRED`
Database User	service-account identifier before `@`

Example service-account mapping:

IAM database user created in Cloud SQL: [email protected]
DBCP Database User: nifi-sa

Verify and Troubleshooting

GcpCloudSqlIamDatabasePasswordProvider Verify checks that NiFi can obtain a token. DBCP Verify checks the
actual database connection using the configured URL, driver, TLS settings, database user, and password provider.

If provider Verify fails:

Confirm the referenced GCPCredentialsControllerService is enabled.
For Workload Identity Federation using service-account impersonation, confirm Target Service Account is set and
the workload identity principal has roles/iam.workloadIdentityUser on that service account.

If provider Verify succeeds but DBCP Verify fails, token acquisition is working and the problem is in the JDBC
connection configuration, network path, TLS settings, driver setup, database user, or database privileges.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Credentials.Service.GCPCredentialsControllerService AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService/additionalDetails.md`)

GCPCredentialsControllerService

The GCPCredentialsControllerService centralizes all authentication strategies used by NiFi components that
interact with Google Cloud. Each strategy exposes only the properties it requires, which lets administrators swap
approaches without touching downstream processors. This guide summarizes every supported strategy.

Application Default Credentials

Application Default Credentials (ADC) allow NiFi to inherit credentials exposed through the runtime environment,
including:

The GOOGLE_APPLICATION_CREDENTIALS environment variable referencing a service-account key file
gcloud auth application-default login on development machines
Cloud Shell or other Google-managed environments that inject ADC automatically

No extra properties are required. Confirm that the account supplying the ADC token has the IAM roles needed by the
processors referencing this controller service.

Service Account Credentials (JSON File)

Use this strategy when the service-account key material is stored on disk. Configure the Service Account JSON
File property to point at the JSON key file. NiFi reads the file when the controller service is enabled and
caches the Google credentials.

Best practices

Restrict filesystem permissions so only the NiFi service user can read the key.
Rotate keys regularly and delete unused keys from the Google Cloud Console.
When impersonating a domain user, set Delegation Strategy to Delegated Account and provide Delegation
User so that NiFi calls Google APIs on behalf of that user.

Service Account Credentials (JSON Value)

This strategy embeds the entire service-account JSON document directly inside the controller-service property. The
value is marked sensitive and can be injected through Parameter Contexts to separate credentials from flow
definitions.

Best practices

Store the JSON value in a Parameter Context referenced by this property so you can swap credentials per
environment.
Use NiFi’s Sensitive Property encryption in nifi.properties to encrypt the stored JSON on disk.

Compute Engine Credentials

Select Compute Engine Credentials when NiFi runs on a Google-managed runtime (Compute Engine, GKE, etc.) and
should use the instance’s attached service account. Google automatically refreshes the metadata server tokens, so
no additional properties are required.

Best practices

Grant the instance service account only the roles required by your flows.
If multiple NiFi nodes share the same instance template, verify that all nodes have access to the same IAM
permissions or configure Workload Identity Federation for finer control.

Workload Identity Federation

Workload Identity Federation (WIF) exchanges an external identity-provider token for a short-lived Google Cloud
access token via Google’s Security Token Service (STS). The controller service configures Google’s
IdentityPoolCredentials, allowing Google client libraries to refresh Google Cloud tokens automatically.

1. Configure Workload Identity Federation in Google Cloud

bash

# Create a pool (only once per environment)
gcloud iam workload-identity-pools create nifi-pool \
  --project=MY_PROJECT_ID \
  --location=global \
  --display-name="NiFi Pool"

# Create a provider bound to your IdP (example for OIDC named myidp)
gcloud iam workload-identity-pools providers create-oidc myidp \
  --project=MY_PROJECT_ID \
  --location=global \
  --workload-identity-pool=nifi-pool \
  --display-name="My Identity Provider" \
  --issuer-uri="https://identity.myidp.com/oauth2/..." \
  --allowed-audiences="//iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/providers/myidp" \
  --attribute-mapping="google.subject=assertion.sub"

Record the audience string printed by the command; it must be copied into NiFi’s Audience property exactly:

text

//iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/providers/myidp

2. Authorize the workload identity principal for Google Cloud resources

When Target Service Account is left blank, the STS-issued access token represents the workload identity principal
itself. Grant IAM roles to that identity on projects or specific resources:

bash

# Project scoped
gcloud projects add-iam-policy-binding MY_PROJECT_ID \
  --member="principal://iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/subject/IDENTITY_SUBJECT" \
  --role="roles/storage.objectViewer"

# Bucket scoped (example)
gcloud storage buckets add-iam-policy-binding gs://MY_BUCKET \
  --member="principal://iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/subject/IDENTITY_SUBJECT" \
  --role="roles/storage.objectViewer"

IDENTITY_SUBJECT must match the claim you mapped in the provider (for example assertion.sub).

When you set Target Service Account, also grant roles/iam.workloadIdentityUser on that service account to the
same workload identity subject so Google can perform service-account impersonation:

bash

gcloud iam service-accounts add-iam-policy-binding \
  "<TARGET_SERVICE_ACCOUNT_EMAIL>" \
  --role="roles/iam.workloadIdentityUser" \
  --member="principal://iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/subject/IDENTITY_SUBJECT"

3. Configure NiFi properties (Workload Identity strategy selected)

Property	Guidance
Audience	Paste the provider resource name recorded above.
Scope	Defaults to `https://www.googleapis.com/auth/cloud-platform`; supply space- or comma-separated scopes if you need fewer permissions.
STS Token Endpoint	Optional override for the Google STS endpoint; leave blank to use `https://sts.googleapis.com/v1/token`.
Subject Token Provider	Controller Service that retrieves the upstream workload identity token (JWT or access token). The token must contain the claims referenced by your attribute mapping.
Subject Token Type	Defaults to `urn:ietf:params:oauth:token-type:jwt`. Choose the alternate access-token type only when the upstream provider issues OAuth access tokens instead of JWTs.
Target Service Account	Optional. Set this only when downstream components must impersonate a Google service account. When configured, the controller service returns generic impersonated credentials and dependent components apply any service-specific scopes when they first use the credential.
Proxy Configuration Service	Optional controller service allowing NiFi to reach STS through HTTP/SOCKS proxies.

Once these properties are set, enable GCPCredentialsControllerService. When Target Service Account is blank,
processors and controller services reference direct IdentityPoolCredentials. When Target Service Account is set,
they reference impersonated credentials layered over the workload identity source credential.

Verification workflow

Enable or refresh the Subject Token Provider controller service.
Use the Verify action on GCPCredentialsControllerService. Successful verification confirms only that NiFi can
construct the configured credential object from the current properties.
Verify does not perform STS exchange, service-account impersonation, or network reachability checks. The first token
refresh happens when a dependent processor or controller service actually uses the credential.
Enable the dependent components that reference this controller service. Product-specific validation happens there.

Troubleshooting

Symptom	Guidance
`403 Caller does not have storage.objects.list`	Confirm the active identity has the required IAM role. Without Target Service Account, grant roles directly to the workload identity principal. With Target Service Account, confirm both the impersonated service account permissions and the `roles/iam.workloadIdentityUser` binding for the workload identity subject.
Verify succeeds but the dependent component fails on first use	Verify only constructs credentials. Recheck the Audience, STS Token Endpoint, upstream subject token, proxy reachability, and any optional Target Service Account setting at the dependent component that triggers the first refresh.
Access token rejected by Google APIs	Revisit IAM bindings and scope selection for the dependent component. If the component requires a service-specific scope or impersonated identity, confirm that Target Service Account is set and that the downstream component supports the needed scope.
Need to rotate upstream tokens	The controller service requests a fresh subject token 60 seconds before expiry. Trigger Refresh on the Subject Token Provider to invalidate cached tokens immediately.

With every strategy available from a single controller-service configuration, NiFi users can migrate between
service-account keys, Compute Engine, Application Default Credentials, and Workload Identity Federation without
introducing new controller services or updating processor properties. Adjust the authentication strategy once and
all dependent processors automatically pick up the new credentials.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Drive.FetchGoogleDrive AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.drive.FetchGoogleDrive/additionalDetails.md`)

FetchGoogleDrive

Accessing Google Drive from NiFi

This processor uses Google Cloud credentials for authentication to access Google Drive. The following steps are required
to prepare the Google Cloud and Google Drive accounts for the processors:

Enable Google Drive API in Google Cloud
- Follow instructions
  at https://developers.google.com/workspace/guides/enable-apis
  and search for 'Google Drive API'.
Grant access to Google Drive folder
- In Google Cloud Console navigate to IAM & Admin -> Service Accounts.
- Take a note of the email of the service account you are going to use.
- Navigate to the folder to be listed in Google Drive.
- Right-click on the Folder -> Share.
- Enter the service account email.
Find File ID
Usually FetchGoogleDrive is used with ListGoogleDrive and 'drive.id' is set.
In case 'drive.id' is not available, you can find the Drive ID of the file in the following way:
- Right-click on the file and select "Get Link".
- In the pop-up window click on "Copy Link".
- You can obtain the file ID from the URL copied to clipboard. For example, if the URL were
  https://drive.google.com/file/d/16ALV9KIU_KKeNG557zyctqy2Fmzyqtq/view?usp=share_link,
  the File ID would be 16ALV9KIU_KKeNG557zyctqy2Fmzyqtq
Set File ID in 'File ID' property

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Drive.ListGoogleDrive AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.drive.ListGoogleDrive/additionalDetails.md`)

ListGoogleDrive

Accessing Google Drive from NiFi

This processor uses Google Cloud credentials for authentication to access Google Drive. The following steps are required
to prepare the Google Cloud and Google Drive accounts for the processors:

Enable Google Drive API in Google Cloud
- Follow instructions
  at https://developers.google.com/workspace/guides/enable-apis
  and search for 'Google Drive API'.
Grant access to Google Drive folder
- In Google Cloud Console navigate to IAM & Admin -> Service Accounts.
- Take a note of the email of the service account you are going to use.
- Navigate to the folder to be listed in Google Drive.
- Right-click on the Folder -> Share.
- Enter the service account email.
Find Folder ID
- Navigate to the folder to be listed in Google Drive and enter it. The URL in your browser will include the ID at
  the end of the URL. For example, if the URL were
  https://drive.google.com/drive/folders/1trTraPVCnX5_TNwO8d9P_bz278xWOmGm, the Folder ID would be
  1trTraPVCnX5_TNwO8d9P_bz278xWOmGm
Set Folder ID in 'Folder ID' property

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Drive.PutGoogleDrive AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.drive.PutGoogleDrive/additionalDetails.md`)

PutGoogleDrive

Accessing Google Drive from NiFi

This processor uses Google Cloud credentials for authentication to access Google Drive. The following steps are required
to prepare the Google Cloud and Google Drive accounts for the processors:

Enable Google Drive API in Google Cloud
- Follow instructions
  at https://developers.google.com/workspace/guides/enable-apis
  and search for 'Google Drive API'.
Grant access to Google Drive folder
- In Google Cloud Console navigate to IAM & Admin -> Service Accounts.
- Take a note of the email of the service account you are going to use.
- Navigate to the folder in Google Drive which will be used as the base folder.
- Right-click on the Folder -> Share.
- Enter the service account email.
Find Folder ID
- Navigate to the folder to be listed in Google Drive and enter it. The URL in your browser will include the ID at
  the end of the URL. For example, if the URL were
  https://drive.google.com/drive/folders/1trTraPVCnX5_TNwO8d9P_bz278xWOmGm, the Folder ID would be
  1trTraPVCnX5_TNwO8d9P_bz278xWOmGm
Set Folder ID in 'Folder ID' property

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Storage.ListGCSBucket AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.storage.ListGCSBucket/additionalDetails.md`)

ListGCSBucket

Streaming Versus Batch Processing

ListGCSBucket performs a listing of all GCS Objects that it encounters in the configured GCS bucket. There are two
common, broadly defined use cases.

Streaming Use Case

By default, the Processor will create a separate FlowFile for each object in the bucket and add attributes for filename,
bucket, etc. A common use case is to connect ListGCSBucket to the FetchGCSObject processor. These two processors used in
conjunction with one another provide the ability to easily monitor a bucket and fetch the contents of any new object as
it lands in GCS in an efficient streaming fashion.

Batch Use Case

To solve this, the ListGCSBucket Processor can optionally be configured with a Record Writer. When a Record Writer is
configured, a single FlowFile will be created that will contain a Record for each object in the bucket, instead of a
separate FlowFile per object. See the documentation for ListFile for an example of how to build a dataflow that allows
for processing all the objects before proceeding with any other step.

One important difference between the data produced by ListFile and ListGCSBucket, though, is the structure of the
Records that are emitted. The Records emitted by ListFile have a different schema than those emitted by ListGCSBucket.
ListGCSBucket emits records that follow the following schema (in Avro format):

text

/* Detailed source-code truncated for AI context efficiency. */

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Vision.GetGcpVisionAnnotateFilesOperationStatus AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.vision.GetGcpVisionAnnotateFilesOperationStatus/additionalDetails.md`)

Google Vision

Get Annotate Files Status

Usage

GetGcpVisionAnnotateFilesOperationStatus is designed to periodically check the statuses of file annotation operations.
This processor should be used in pair with StartGcpVisionAnnotateFilesOperation Processor. An outgoing FlowFile contains
the raw response returned by the Vision server. This response is in JSON format and contains a Google storage reference
where the result is located, as well as additional metadata, as written in
the Google Vision API Reference document.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Vision.GetGcpVisionAnnotateImagesOperationStatus AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.vision.GetGcpVisionAnnotateImagesOperationStatus/additionalDetails.md`)

Google Vision

Google Cloud Vision - Get Annotate Images Status

Usage

GetGcpVisionAnnotateImagesOperationStatus is designed to periodically check the statuses of image annotation operations.
This processor should be used in pair with StartGcpVisionAnnotateImagesOperation Processor. An outgoing FlowFile
contains the raw response returned by the Vision server. This response is in JSON format and contains a Google storage
reference where the result is located, as well as additional metadata, as written in
the Google Vision API Reference document.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Vision.StartGcpVisionAnnotateFilesOperation AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.vision.StartGcpVisionAnnotateFilesOperation/additionalDetails.md`)

Google Vision

Google Cloud Vision - Start Annotate Files Operation

Prerequisites

Make sure Vision API is enabled and the account you are using has the right to use it
Make sure the input file(s) are available in a GCS bucket

Usage

StartGcpVisionAnnotateFilesOperation is designed to trigger file annotation operations. This processor should be used in
pair with the GetGcpVisionAnnotateFilesOperationStatus Processor. Outgoing FlowFiles contain the raw response to the
request returned by the Vision server. The response is in JSON format and contains the result and additional metadata as
written in the Google Vision API Reference documents.

Payload

The JSON Payload is a request in JSON format as documented in
the Google Vision REST API reference document.
Payload can be fed to the processor via the JSON Payload property or as a FlowFile content. The property has higher
precedence over FlowFile content. Please make sure to delete the default value of the property if you want to use
FlowFile content payload. A JSON payload template example:

json

{
  "requests": [
    {
      "inputConfig": {
        "gcsSource": {
          "uri": "gs://${gcs.bucket}/${filename}"
        },
        "mimeType": "application/pdf"
      },
      "features": [
        {
          "type": "${vision-feature-type}",
          "maxResults": 4
        }
      ],
      "outputConfig": {
        "gcsDestination": {
          "uri": "gs://${output-bucket}/${filename}/"
        },
        "batchSize": 2
      }
    }
  ]
}

Features types

TEXT_DETECTION: Optical character recognition (OCR) for an image; text recognition and conversion to machine-coded
text. Identifies and extracts UTF-8 text in an image.
DOCUMENT_TEXT_DETECTION: Optical character recognition (OCR) for a file (PDF/TIFF) or dense text image; dense text
recognition and conversion to machine-coded text.

You can find more details at Google Vision Feature List

Example: How to set up a simple Annotate Image Flow

Prerequisites

Create an input and output bucket
Input files should be available in a GCS bucket
This bucket must not contain anything else but the input files
Set the bucket property of ListGCSBucket processor to your input bucket name
Keep the default value of JSON PAYLOAD property in StartGcpVisionAnnotateFilesOperation
Set the Output Bucket property to your output bucket name in StartGcpVisionAnnotateFilesOperation
Setup GCP Credentials Provider Service for all GCP related processor

Execution steps:

ListGCSBucket processor will return a list of files in the bucket at the first run.
ListGCSBucket will return only new items at subsequent runs.
StartGcpVisionAnnotateFilesOperation processor will trigger GCP Vision file annotation jobs based on the JSON payload.
StartGcpVisionAnnotateFilesOperation processor will populate the operationKey flow file attribute.
GetGcpVisionAnnotateFilesOperationStatus processor will periodically query status of the job.

Nifi Extension Bundles Nifi Gcp Bundle Nifi Gcp Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Gcp.Vision.StartGcpVisionAnnotateImagesOperation AdditionalDetails (`nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.vision.StartGcpVisionAnnotateImagesOperation/additionalDetails.md`)

Google Vision

Google Cloud Vision - Start Annotate Images Operation

Prerequisites

Make sure Vision API is enabled and the account you are using has the right to use it
Make sure the input image(s) are available in a GCS bucket under /input folder

Usage

StartGcpVisionAnnotateImagesOperation is designed to trigger image annotation operations. This processor should be used
in pair with the GetGcpVisionAnnotateImagesOperationStatus Processor. Outgoing FlowFiles contain the raw response to the
request returned by the Vision server. The response is in JSON format and contains the result and additional metadata as
written in the Google Vision API Reference documents.

Payload

json

{
  "requests": [
    {
      "image": {
        "source": {
          "imageUri": "gs://${gcs.bucket}/${filename}"
        }
      },
      "features": [
        {
          "type": "${vision-feature-type}",
          "maxResults": 4
        }
      ]
    }
  ],
  "outputConfig": {
    "gcsDestination": {
      "uri": "gs://${output-bucket}/${filename}/"
    },
    "batchSize": 2
  }
}

Features types

TEXT_DETECTION: Optical character recognition (OCR) for an image; text recognition and conversion to machine-coded
text. Identifies and extracts UTF-8 text in an image.
DOCUMENT_TEXT_DETECTION: Optical character recognition (OCR) for a file (PDF/TIFF) or dense text image; dense text
recognition and conversion to machine-coded text.
LANDMARK_DETECTION: Provides the name of the landmark, a confidence score and a bounding box in the image for the
landmark.
LOGO_DETECTION: Provides a textual description of the entity identified, a confidence score, and a bounding polygon
for the logo in the file.
LABEL_DETECTION: Provides generalized labels for an image.
etc.

You can find more details at Google Vision Feature List

Example: How to set up a simple Annotate Image Flow

Prerequisites

Create an input and output bucket
Input image files should be available in a GCS bucket
This bucket must not contain anything else but the input image files
Set the bucket property of ListGCSBucket processor to your input bucket name
Keep the default value of JSON PAYLOAD property in StartGcpVisionAnnotateImagesOperation
Set the Output Bucket property to your output bucket name in StartGcpVisionAnnotateImagesOperation
Setup GCP Credentials Provider Service for all GCP related processor

Execution steps:

ListGCSBucket processor will return a list of files in the bucket at the first run.
ListGCSBucket will return only new items at subsequent runs.
StartGcpVisionAnnotateImagesOperation processor will trigger GCP Vision image annotation jobs based on the JSON
payload.
StartGcpVisionAnnotateImagesOperation processor will populate the operationKey flow file attribute.
GetGcpVisionAnnotateImagesOperationStatus processor will periodically query status of the job.

Nifi Extension Bundles Nifi Geohash Bundle Nifi Geohash Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Geohash.GeohashRecord AdditionalDetails (`nifi-extension-bundles/nifi-geohash-bundle/nifi-geohash-processors/src/main/resources/docs/org.apache.nifi.processors.geohash.GeohashRecord/additionalDetails.md`)

GeohashRecord

Overview

A Geohash value corresponds to a specific area with pre-defined granularity and is widely used in identifying,
representing and indexing geospatial objects. This GeohashRecord processor provides the ability to encode and decode
Geohashes with desired format and precision.

Formats supported

BASE32: The most commonly used alphanumeric version. It is compact and more human-readable by discarding some letters
(such as "a" and "o", "i" and "l") that might cause confusion.
BINARY: This format is generated by directly interleaving latitude and longitude binary strings. The even bits in the
binary strings correspond to the longitude, while the odd digits correspond to the latitude.
LONG: Although this 64-bit number format is not human-readable, it can be calculated very fast and is more efficient.

Precision supported

In ENCODE mode, users specify the desired precision level, which should be an integer number between 1 and 12. A
greater level will generate a longer Geohash with higher precision.

In DECODE mode, users are not asked to provide a precision level because this information is contained in the length of
Geohash values given.

Nifi Extension Bundles Nifi Graph Bundle Nifi Graph Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Graph.ExecuteGraphQuery AdditionalDetails (`nifi-extension-bundles/nifi-graph-bundle/nifi-graph-processors/src/main/resources/docs/org.apache.nifi.processors.graph.ExecuteGraphQuery/additionalDetails.md`)

ExecuteGraphQuery

This processor is designed to work with Gremlin and Cypher queries. The query is specified in the configuration
parameter labeled Query, and parameters can be configured using dynamic properties on the processor. All Gremlin and
Cypher CRUD operations are supported by this processor. It will stream the entire result set into a single flowfile as a
JSON array.

More Information

Nifi Extension Bundles Nifi Graph Bundle Nifi Other Graph Services Src Main Resources Docs Org.Apache.Nifi.Graph.TinkerpopClientService AdditionalDetails (`nifi-extension-bundles/nifi-graph-bundle/nifi-other-graph-services/src/main/resources/docs/org.apache.nifi.graph.TinkerpopClientService/additionalDetails.md`)

GremlinClientService

This client service configures a connection to a Gremlin Server and allows Gremlin queries to be executed against the
Gremlin Server. For more information on Gremlin and Gremlin Server, see
the Apache Tinkerpop project.

This client service supports two differnt modes of operation: Script Submission and Bytecode Submission, described
below.

Script Submission

Script submission is the default way to interact with the gremlin server. This takes the input script and
uses Script Submission to interact with the
gremlin server. Because the script is shipped to the gremlin server as a string, only simple queries are recommended (
count, path, etc.) as there are no complex serializers available in this operation. This also means that NiFi will not
be opinionated about what is returned, whatever the response from the tinkerpop server is, the response will be
deserialized assuming common Java types. In the case of a Map return, the values will be returned as a record in the
FlowFile response, in all other cases, the return of the query will be coerced into a Map with key "result" and value
being the result of your script submission for that specific response.

Serialization Issues in Script Submission

A common issue when creating Gremlin scripts for first time users is to accidentally return an unserializable object.
Gremlin is a Groovy DSL and so it behaves like compiled Groovy including returning the last statement in the script.
This is an example of a Gremlin script that could cause unexpected failures:

text

g.V().hasLabel("person").has("name", "John Smith").valueMap()

The valueMap() step is not directly serializable and will fail. To fix that you have two potential options:

text

//Return a Map
g.V().hasLabel("person").has("name", "John Smith").valueMap().next()

Alternative:

text

g.V().hasLabel("person").has("name", "John Smith").valueMap()
true //Return boolean literal

Bytecode Submission

Bytecode submission is the more flexible of the two submission method and will be much more performant in a production
system. When combined with the Yaml connection settings and a custom jar, very complex graph queries can be run directly
within the NiFi JVM, leveraging custom serializers to decrease serialization overhead.

Instead of submitting a script to the gremlin server, requiring string serialization on both sides of the string result
set, the groovy script is compiled within the NiFi JVM. This compiled script has the bindings of g (the
GraphTraversalSource) and log (the NiFi logger) injected into the compiled code. Utilizing g, your result set is
contained within NiFi and serialization should take care of the overhead of your responses drastically decreasing the
likelihood of serialization errors.

As the result returned cannot be known by NiFi to be a specific type, your groovy script must rerun a Map<String,
Object>, otherwise the response will be ignored. Here is an example:

text

Object results = g.V().hasLabel("person").has("name", "John Smith").valueMap().collect()
[result: results]

This will break up your response objects into an array within your result key, allowing further processing within nifi
if necessary.

Nifi Extension Bundles Nifi Groovyx Bundle Nifi Groovyx Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Groovyx.ExecuteGroovyScript AdditionalDetails (`nifi-extension-bundles/nifi-groovyx-bundle/nifi-groovyx-processors/src/main/resources/docs/org.apache.nifi.processors.groovyx.ExecuteGroovyScript/additionalDetails.md`)

Groovy

Summary

This is the grooviest groovy script :)

Script Bindings:

variable	type	description
session	org.apache.nifi.processor.ProcessSession	the session that is used to get, change, and transfer input files
context	org.apache.nifi.processor.ProcessContext	the context (almost unuseful)
log	org.apache.nifi.logging.ComponentLog	the logger for this processor instance
REL_SUCCESS	org.apache.nifi.processor.Relationship	the success relationship
REL_FAILURE	org.apache.nifi.processor.Relationship	the failure relationship
CTL	java.util.HashMap<String,ControllerService>	Map populated with controller services defined with `CTL.*` processor properties. The `CTL.` prefixed properties could be linked to controller service and provides access to this service from a script without additional code.
SQL	java.util.HashMap<String, groovy.sql.Sql>	Map populated with `groovy.sql.Sql` objects connected to corresponding database defined with `SQL.*` processor properties. The `SQL.` prefixed properties could be linked only to DBCPSercice.
RecordReader	java.util.HashMap<String,RecordReaderFactory>	Map populated with controller services defined with `RecordReader.*` processor properties. The `RecordReader.` prefixed properties are to be linked to RecordReaderFactory controller service instances.
RecordWriter	java.util.HashMap<String,RecordSetWriterFactory>	Map populated with controller services defined with `RecordWriter.*` processor properties. The `RecordWriter.` prefixed properties are to be linked to RecordSetWriterFactory controller service instances.
Dynamic processor properties	org.apache.nifi.components.PropertyDescriptor	All processor properties not started with `CTL.` or `SQL.` are bound to script variables

SQL map details

Example: if you defined property `SQL.mydb` and linked it to any DBCPService, then you can access it from code
SQL.mydb.rows('select * from mytable')

The processor automatically takes connection from dbcp service before executing script and tries to handle
transaction:
database transactions automatically rolled back on script exception and committed on success.
Or you can manage transaction manually.
NOTE: Script must not disconnect connection.

SessionFile - flow file extension

The (org.apache.nifi.processors.groovyx.flow.SessionFile) is an actual object returned by session in Extended Groovy
processor.
This flow file is a container that references session and the real flow file.
This allows to use simplified syntax to work with file attributes and content:

set new attribute value

groovy

flowFile.ATTRIBUTE_NAME = ATTRIBUTE_VALUE
flowFile.'mime.type' = 'text/xml'
flowFile.putAttribute("ATTRIBUTE_NAME", ATTRIBUTE_VALUE)
// the same as
flowFile = session.putAttribute(flowFile, "ATTRIBUTE_NAME", ATTRIBUTE_VALUE)

remove attribute

groovy

flowFile.ATTRIBUTE_NAME = null
// equals to
flowFile = session.removeAttribute(flowFile, "ATTRIBUTE_NAME")

get attribute value

groovy

String a = flowFile.ATTRIBUTE_NAME

write content

groovy

flowFile.write("UTF-8", "THE CharSequence to write into flow file replacing current content")
flowFile.write("UTF-8") { writer ->
    // do something with java.io.Writer...
}
flowFile.write { outStream ->
    // do something with output stream...
}
flowFile.write { inStream, outStream ->
    // do something with input and output streams...
}

get content

groovy

InputStream i = flowFile.read()
def json = new groovy.json.JsonSlurper().parse(flowFile.read())
String text = flowFile.read().getText("UTF-8")

transfer flow file to success relation

groovy

REL_SUCCESS << flowFile
flowFile.transfer(REL_SUCCESS)
// the same as:
session.transfer(flowFile, REL_SUCCESS)

work with dbcp

groovy

import groovy.sql.Sql

// define property named `SQL.db` connected to a DBCPConnectionPool controller service
// for this case it's an H2 database example

// read value from the database with prepared statement
// and assign into flowfile attribute `db.yesterday`
def daysAdd = -1
def row = SQL.db.firstRow("select dateadd('DAY', ${daysAdd}, sysdate) as DB_DATE from dual")
flowFile.'db.yesterday' = row.DB_DATE

// to work with BLOBs and CLOBs in the database
// use parameter casting using groovy.sql.Sql.BLOB(Stream) and groovy.sql.Sql.CLOB(Reader)

// write content of the flow file into database blob
flowFile.read { rawIn ->
    def parms = [
            p_id  : flowFile.ID as Long, // get flow file attribute named \`ID\`
            p_data: Sql.BLOB(rawIn),   // use input stream as BLOB sql parameter
    ]
    SQL.db.executeUpdate(parms, "update mytable set data = :p_data where id = :p_id")
}

Handling processor start & stop

In the extended groovy processor you can catch `start` and `stop` and `unscheduled` events by providing
corresponding static methods:

groovy

import org.apache.nifi.processor.ProcessContext
import java.util.concurrent.atomic.AtomicLong

class Const {
    static Date startTime = null;
    static AtomicLong triggerCount = null;
}

static onStart(ProcessContext context) {
    Const.startTime = new Date()
    Const.triggerCount = new AtomicLong(0)
    println "onStart $context ${Const.startTime}"
}

static onStop(ProcessContext context) {
    def alive = (System.currentTimeMillis() - Const.startTime.getTime()) / 1000
    println "onStop $context executed ${Const.triggerCount} times during ${alive} seconds"
}

static onUnscheduled(ProcessContext context) {
    def alive = (System.currentTimeMillis() - Const.startTime.getTime()) / 1000
    println "onUnscheduled $context executed ${Const.triggerCount} times during ${alive} seconds"
}

flowFile.'trigger.count' = Const.triggerCount.incrementAndGet()
REL_SUCCESS << flowFile

Nifi Extension Bundles Nifi Hadoop Bundle Nifi Hdfs Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Hadoop.CreateHadoopSequenceFile AdditionalDetails (`nifi-extension-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/resources/docs/org.apache.nifi.processors.hadoop.CreateHadoopSequenceFile/additionalDetails.md`)

CreateHadoopSequenceFile

Description

This processor is used to create a Hadoop Sequence File, which essentially is a file of key/value pairs. The key will be
a file name and the value will be the flow file content. The processor will take either a merged (a.k.a. packaged) flow
file or a singular flow file. Historically, this processor handled the merging by type and size or time prior to
creating a SequenceFile output; it no longer does this. If creating a SequenceFile that contains multiple files of the
same type is desired, precede this processor with a RouteOnAttribute processor to segregate files of the same type and
follow that with a MergeContent processor to bundle up files. If the type of files is not important, just use the
MergeContent processor. When using the MergeContent processor, the following Merge Formats are supported by this
processor:

TAR
ZIP
FlowFileStream v3

The created SequenceFile is named the same as the incoming FlowFile with the suffix '.sf'. For incoming FlowFiles that
are bundled, the keys in the SequenceFile are the individual file names, the values are the contents of each file.

NOTE: The value portion of a key/value pair is loaded into memory. While there is a max size limit of 2GB, this could
cause memory issues if there are too many concurrent tasks and the flow file sizes are large.

Using Compression

The value of the Compression codec property determines the compression library the processor uses to compress content.
Third party libraries are used for compression. These third party libraries can be Java libraries or native libraries.
In case of native libraries, the path of the parent folder needs to be in an environment variable called
LD_LIBRARY_PATH so that NiFi can find the libraries.

Example: using Snappy compression with native library on CentOS

Snappy compression needs to be installed on the server running NiFi:
sudo yum install snappy
Suppose that the server running NiFi has the native compression libraries in /opt/lib/hadoop/lib/native . (Native
libraries have file extensions like .so, .dll, .lib, etc. depending on the platform.)
We need to make sure that the files can be executed by the NiFi process' user. For this purpose we can make a copy of
these files to e.g. /opt/nativelibs and change their owner. If NiFi is executed by nifi user in the nifi group,
then:
chown nifi:nifi /opt/nativelibs
chown nifi:nifi /opt/nativelibs/*
The LD_LIBRARY_PATH needs to be set to contain the path to the folder /opt/nativelibs.
NiFi needs to be restarted.
Compression codec property can be set to SNAPPY and a Compression type can be selected.
The processor can be started.

Nifi Extension Bundles Nifi Hadoop Bundle Nifi Hdfs Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Hadoop.ListHDFS AdditionalDetails (`nifi-extension-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/resources/docs/org.apache.nifi.processors.hadoop.ListHDFS/additionalDetails.md`)

ListHDFS

ListHDFS Filter Modes

There are three filter modes available for ListHDFS that determine how the regular expression in the File Filter
property will be applied to listings in HDFS.

Directories and Files
Filtering will be applied to the names of directories and files. If Recurse Subdirectories is set to true, only
subdirectories with a matching name will be searched for files that match the regular expression defined in
File Filter.
Files Only
Filtering will only be applied to the names of files. If Recurse Subdirectories is set to true, the entire
subdirectory tree will be searched for files that match the regular expression defined in File Filter.
Full Path
Filtering will be applied to the full path of files. If Recurse Subdirectories is set to true, the entire
subdirectory tree will be searched for files in which the full path of the file matches the regular expression defined
in File Filter. Regarding scheme and authority, if a given file has a full path of
hdfs://hdfscluster:8020/data/txt/1.txt, the filter will evaluate the regular expression defined in File Filter
against two cases, matching if either is true:
the full path including the scheme (hdfs), authority (hdfscluster:8020), and the remaining path components (
/data/txt/1.txt)
only the path components (/data/txt/1.txt)

Examples:

For the given examples, the following directory structure is used:

data
├── readme.txt
├── bin
│ ├── readme.txt
│ ├── 1.bin
│ ├── 2.bin
│ └── 3.bin
├── csv
│ ├── readme.txt
│ ├── 1.csv
│ ├── 2.csv
│ └── 3.csv
└── txt ├── readme.txt ├── 1.txt ├── 2.txt └── 3.txt

Directories and Files

This mode is useful when the listing should match the names of directories and files with the regular expression defined
in File Filter. When Recurse Subdirectories is true, this mode allows the user to filter for files in
subdirectories with names that match the regular expression defined in File Filter.

ListHDFS configuration:

Property	Value
`Directory`	`/data`
`Recurse Subdirectories`	true
`File Filter`	`.txt.`
`Filter Mode`	`Directories and Files`

ListHDFS results:

/data/readme.txt
/data/txt/readme.txt
/data/txt/1.txt
/data/txt/2.txt
/data/txt/3.txt

Files Only

This mode is useful when the listing should match only the names of files with the regular expression defined in
File Filter. Directory names will not be matched against the regular expression defined in File Filter. When
Recurse Subdirectories is true, this mode allows the user to filter for files in the entire subdirectory tree of
the directory specified in the Directory property.

ListHDFS configuration:

Property	Value
`Directory`	`/data`
`Recurse Subdirectories`	true
`File Filter`	`[^\.].*\.txt`
`Filter Mode`	`Files Only`

ListHDFS results:

/data/readme.txt
/data/bin/readme.txt
/data/csv/readme.txt
/data/txt/readme.txt
/data/txt/1.txt
/data/txt/2.txt
/data/txt/3.txt

Full Path

This mode is useful when the listing should match the entire path of a file with the regular expression defined in
File Filter. When Recurse Subdirectories is true, this mode allows the user to filter for files in the entire
subdirectory tree of the directory specified in the Directory property while allowing filtering based on the full
path of each file.

ListHDFS configuration:

Property	Value
`Directory`	`/data`
`Recurse Subdirectories`	true
`File Filter`	`(/./)csv/.*`
`Filter Mode`	`Full Path`

ListHDFS results:

/data/csv/readme.txt
/data/csv/1.csv
/data/csv/2.csv
/data/csv/3.csv

Streaming Versus Batch Processing

ListHDFS performs a listing of all files that it encounters in the configured HDFS directory. There are two common,
broadly defined use cases.

Streaming Use Case

By default, the Processor will create a separate FlowFile for each file in the directory and add attributes for
filename, path, etc. A common use case is to connect ListHDFS to the FetchHDFS processor. These two processors used in
conjunction with one another provide the ability to easily monitor a directory and fetch the contents of any new file as
it lands in HDFS in an efficient streaming fashion.

Batch Use Case

Another common use case is the desire to process all newly arriving files in a given directory, and to then perform some
action only when all files have completed their processing. The above approach of streaming the data makes this
difficult, because NiFi is inherently a streaming platform in that there is no "job" that has a beginning and an end.
Data is simply picked up as it becomes available.

To solve this, the ListHDFS Processor can optionally be configured with a Record Writer. When a Record Writer is
configured, a single FlowFile will be created that will contain a Record for each file in the directory, instead of a
separate FlowFile per file. See the documentation for ListFile for an example of how to build a dataflow that allows for
processing all the files before proceeding with any other step.

One important difference between the data produced by ListFile and ListHDFS, though, is the structure of the Records
that are emitted. The Records emitted by ListFile have a different schema than those emitted by ListHDFS. ListHDFS emits
records that follow the following schema (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    {
      "name": "filename",
      "type": "string"
    },
    {
      "name": "path",
      "type": "string"
    },
    {
      "name": "directory",
      "type": "boolean"
    },
    {
      "name": "size",
      "type": "long"
    },
    {
      "name": "lastModified",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "permissions",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "owner",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "group",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "replication",
      "type": [
        "null",
        "int"
      ]
    },
    {
      "name": "symLink",
      "type": [
        "null",
        "boolean"
      ]
    },
    {
      "name": "encrypted",
      "type": [
        "null",
        "boolean"
      ]
    },
    {
      "name": "erasureCoded",
      "type": [
        "null",
        "boolean"
      ]
    }
  ]
}

Nifi Extension Bundles Nifi Hadoop Bundle Nifi Hdfs Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Hadoop.PutHDFS AdditionalDetails (`nifi-extension-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/resources/docs/org.apache.nifi.processors.hadoop.PutHDFS/additionalDetails.md`)

PutHDFS

SSL Configuration:

Hadoop provides the ability to configure keystore and/or truststore properties. If you want to use SSL-secured file
system like swebhdfs, you can use the Hadoop configurations instead of using SSL Context Service.

create 'ssl-client.xml' to configure the truststores.

ssl-client.xml Properties:

Property	Default Value	Explanation
ssl.client.truststore.type	jks	Truststore file type
ssl.client.truststore.location	NONE	Truststore file location
ssl.client.truststore.password	NONE	Truststore file password
ssl.client.truststore.reload.interval	10000	Truststore reload interval, in milliseconds

ssl-client.xml Example:

xml

<configuration>
    <property>
        <name>ssl.client.truststore.type</name>
        <value>jks</value>
    </property>
    <property>
        <name>ssl.client.truststore.location</name>
        <value>/path/to/truststore.jks</value>
    </property>
    <property>
        <name>ssl.client.truststore.password</name>
        <value>clientfoo</value>
    </property>
    <property>
        <name>ssl.client.truststore.reload.interval</name>
        <value>10000</value>
    </property>
</configuration>

put 'ssl-client.xml' to the location looked up in the classpath, like under NiFi conriguration directory.
set the name of 'ssl-client.xml' to hadoop.ssl.client.conf in the 'core-site.xml' which HDFS processors use.

xml

<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>swebhdfs://{namenode.hostname:port}</value>
    </property>
    <property>
        <name>hadoop.ssl.client.conf</name>
        <value>ssl-client.xml</value>
    </property>
</configuration>

Nifi Extension Bundles Nifi Hashicorp Vault Bundle Nifi Hashicorp Vault Client Service Src Main Resources Docs Org.Apache.Nifi.Vault.Hashicorp.StandardHashiCorpVaultClientService AdditionalDetails (`nifi-extension-bundles/nifi-hashicorp-vault-bundle/nifi-hashicorp-vault-client-service/src/main/resources/docs/org.apache.nifi.vault.hashicorp.StandardHashiCorpVaultClientService/additionalDetails.md`)

StandardHashiCorpVaultClientService

Configuring the Bootstrap HashiCorp Vault Configuration File

The ./conf/bootstrap-hashicorp-vault.conf file that comes with Apache NiFi is a convenient way to configure this
controller service in a manner consistent with the HashiCorpVault sensitive property provider. Since this file is
already used for configuring the Vault client for protecting sensitive properties in the NiFi configuration files (see
the Administrator's Guide), it's a natural starting point for configuring the controller service as well.

An example configuration of this properties file is as follows:

properties

# HTTP or HTTPS URI for HashiCorp Vault is required to enable the Sensitive Properties Provider
vault.uri=https://127.0.0.1:8200

# Optional file supports authentication properties described in the Spring Vault Environment Configuration
# https://docs.spring.io/spring-vault/docs/2.3.x/reference/html/#vault.core.environment-vault-configuration
#
# All authentication properties must be included in bootstrap-hashicorp-vault.conf when this property is not specified.
# Properties in bootstrap-hashicorp-vault.conf take precedence when the same values are defined in both files.
# Token Authentication is the default when the 'vault.authentication' property is not specified.
vault.authentication.properties.file=[full/path/to/vault-auth.properties]

# Optional Timeout properties
vault.connection.timeout=5 secs
vault.read.timeout=15 secs

# Optional TLS properties
vault.ssl.enabledCipherSuites=
vault.ssl.enabledProtocols=TLSv1.3
vault.ssl.key-store=[path/to/keystore.p12]
vault.ssl.key-store-type=PKCS12
vault.ssl.key-store-password=[keystore password]
vault.ssl.trust-store=[path/to/truststore.p12]
vault.ssl.trust-store-type=PKCS12
vault.ssl.trust-store-password=[truststore password]

In order to use this file in the StandardHashiCorpVaultClientService, specify the following properties:

Configuration Strategy - Properties Files
Vault Properties Files - ./conf/bootstrap-hashicorp-vault.conf

If your bootstrap configuration includes the vault.authentication.properties.file containing additional authentication
properties, this file will also need to be added to the Vault Properties Files property as a comma-separated value.

Configuring the Client using Direct Properties

However, if you want to specify or override properties directly in the controller service, you may do this by specifying
a Configuration Strategy of 'Direct Properties'. This can be useful if you are reusing an SSLContextService or want to
parameterize the Vault configuration properties. Authentication-related properties can also be added as sensitive
dynamic properties, as seen in the examples below.

Vault Authentication

Under the hood, the controller service uses Spring Vault, and directly supports the property keys specified
in Spring Vault's documentation.
Following are some common examples of authentication with Vault.

Token Authentication

The simplest authentication scheme uses a rotating token, which is enabled by default in Vault. To specify this
mechanism, select "TOKEN" from the "Vault Authentication" property (the default). However, since the token should rotate
by nature, it is a best practice to use the 'Properties Files' Configuration Strategy, and keep the token value in an
external properties file, indicating this filename in the 'Vault Properties Files' property. Then an external process
can rotate the token in the file without updating NiFi configuration. In order to pick up the changed token, the
controller service must be disabled and re-enabled.

For testing purposes, however, it may be more convenient to specify the token directly in the controller service. To do
so, add a new Sensitive property named 'vault.token' and enter the token as the value.

Certificate Authentication

Certificate authentication must be enabled in the Vault server before it can be used from NiFi, but it uses the same TLS
settings as the actual client connection, so no additional authentication properties are required. While these TLS
settings can be provided in an external properties file, we will demonstrate configuring an SSLContextService instead.

First, create an SSLContextService controller service and configure the Filename, Password, and Type for both the
Keystore and Truststore. Enable it, and assign it as the SSL Context Service in the Vault controller service. Then,
simply specify "CERT" as the "Vault Authentication" property value.

Other Authentication Methods

To configure the other authentication methods, see the Spring Vault documentation linked above. All relevant properties
should be added either to the external properties files referenced in the "Vault Properties Files" property if using
the 'Properties Files' Configuration Strategy, or added as custom properties with the same name if using the 'Direct
Properties' Configuration Strategy. For example, for the Azure authentication mechanism, properties will have to be
added for 'vault.azure-msi.azure-path', 'vault.azure-msi.role', and 'vault.azure-msi.identity-token-service'.

Nifi Extension Bundles Nifi Hazelcast Bundle Nifi Hazelcast Services Src Main Resources Docs Org.Apache.Nifi.Hazelcast.Services.Cacheclient.HazelcastMapCacheClient AdditionalDetails (`nifi-extension-bundles/nifi-hazelcast-bundle/nifi-hazelcast-services/src/main/resources/docs/org.apache.nifi.hazelcast.services.cacheclient.HazelcastMapCacheClient/additionalDetails.md`)

HazelcastMapCacheClient

This implementation of distributed map cache is backed by Hazelcast. The Hazelcast connection is provided and maintained
by an instance of HazelcastCacheManager. One HazelcastCacheManager might serve multiple cache clients. This
implementation uses the IMap data structure. The identifier of the Hazelcast IMap will be the same as the value of the
property Hazelcast Cache Name. It is recommended for all HazelcastMapCacheClient instances to use different cache names.

The implementation supports the atomic method family defined in AtomicDistributedMapCacheClient. This is achieved by
maintaining a revision number for every entry. The revision is an 8 byte long integer. It is increased when the entry is
updated. The value is kept during modifications not part of the atomic method family but this is mainly for regular
management of the entries. It is not recommended to work with elements by mixing the two method families.

The convention for all the entries is to reserve the first 8 bytes for the revision. The rest of the content is the
serialized payload.

Nifi Extension Bundles Nifi Hazelcast Bundle Nifi Hazelcast Services Src Main Resources Docs Org.Apache.Nifi.Hazelcast.Services.Cachemanager.EmbeddedHazelcastCacheManager AdditionalDetails (`nifi-extension-bundles/nifi-hazelcast-bundle/nifi-hazelcast-services/src/main/resources/docs/org.apache.nifi.hazelcast.services.cachemanager.EmbeddedHazelcastCacheManager/additionalDetails.md`)

EmbeddedHazelcastCacheManager

This service starts and manages an embedded Hazelcast instance. The cache manager has direct accesses to the instance -
and the data stored in it. However, the instance opens a port for potential clients to join and this cannot be
prevented. Note that this might leave the instance open for rogue clients to join.

It is possible to have multiple independent Hazelcast instances on the same host (whether via
EmbeddedHazelcastCacheManager or externally) without any interference by setting the properties accordingly. If there
are no other instances, the default cluster name and port number can simply be used.

The service supports multiple ways to set up a Hazelcast cluster. This is controlled by the property, named "Hazelcast
Clustering Strategy". The following strategies may be used:

None

This is the default value. Used when sharing data between nodes is not required. With this value, every NiFi node in the
cluster (if it is clustered) connects to its local Hazelcast server only. The Hazelcast servers do not form a cluster.

All Nodes

Can be used only in clustered node. Using this strategy will result a single Hazelcast cluster consisting of the
embedded instances of all the NiFi nodes. This strategy requires all Hazelcast servers listening on the same port.
Having different port numbers (based on expression for example) would prevent the cluster from forming.

The controller service automatically gathers the host list from the NiFi cluster itself when it is enabled. It is not
required for all the nodes to have been successfully joined at this point, but the join must have been initiated. When
the controller service is enabled at the start of the NiFi instance, the enabling of the service will be prevented until
the node is considered clustered.

Hazelcast can accept nodes that join at a later time. As the new node has a comprehensive list of the expected
instances - including the already existing ones and itself - Hazelcast will be able to reach the expected state. Beware:
this may take significant time.

Explicit

Can be used only in clustered node. Explicit Clustering Strategy allows more control over the Hazelcast cluster members.
All Nodes Clustering Strategy, this strategy works with a list of Hazelcast servers, but instance discovery is not
automatic.

This strategy uses property named "Hazelcast Instances" to determine the members of the Hazelcast clusters. This list of
hosts must contain all the instances expected to be part of the cluster. The instance list may contain only hosts which
are part of the NiFi cluster. The port specified in the "Hazelcast Port" will be used as Hazelcast server port.

In case the current node is not part of the instance list, the service will start a Hazelcast client. The client then
will connect to the Hazelcast addresses specified in the instance list. Users of the service should not perceive any
difference in functionality.

Nifi Extension Bundles Nifi Hazelcast Bundle Nifi Hazelcast Services Src Main Resources Docs Org.Apache.Nifi.Hazelcast.Services.Cachemanager.ExternalHazelcastCacheManager AdditionalDetails (`nifi-extension-bundles/nifi-hazelcast-bundle/nifi-hazelcast-services/src/main/resources/docs/org.apache.nifi.hazelcast.services.cachemanager.ExternalHazelcastCacheManager/additionalDetails.md`)

ExternalHazelcastCacheManager

This service connects to an external Hazelcast cluster (or standalone instance) as client. Hazelcast 4.0.0 or newer
version is required. The connection to the server is kept alive using Hazelcast's built-in reconnection capability. This
might be fine-tuned by setting the following properties:

Hazelcast Initial Backoff
Hazelcast Maximum Backoff
Hazelcast Backoff Multiplier
Hazelcast Connection Timeout

If the service cannot connect or abruptly disconnected it tries to reconnect after a backoff time. The amount of time
waiting before the first attempt is defined by the Initial Backoff. If the connection is still not successful the client
waits gradually more between the attempts until the waiting time reaches the value set in the 'Hazelcast Maximum
Backoff' property (or the connection timeout, whichever is smaller). The backoff time after the first attempt is always
based on the previous amount, multiplied by the Backoff Multiplier. Note: the real backoff time might be slightly differ
as some "jitter" is added to the calculation in order to avoid regularity.

Nifi Extension Bundles Nifi Hubspot Bundle Nifi Hubspot Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Hubspot.GetHubSpot AdditionalDetails (`nifi-extension-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/main/resources/docs/org.apache.nifi.processors.hubspot.GetHubSpot/additionalDetails.md`)

GetHubSpot

Authentication Methods

The processor is working with HubSpot private applications. A HubSpot private app must be created (
see HubSpot Private App Creation) in order to connect to HubSpot
and make requests. Private App Access Tokens are the only authentication method that is currently supported.

Incremental Loading

HubSpot objects can be processed incrementally by NiFi. This means that only objects created or modified after the last
run time of the processor are processed. The processor state can be reset in the context menu. The incremental loading
is based on the objects last modified time.

Paging

GetHubSpot supports both paging and incrementality at the same time. In case the number of results exceeds the 'Result
Limit', in the next processor run the remaining objects will be returned.

Due to the page handling mechanism of the HubSpot API, parallel deletions are not supported. Some objects may be omitted
if any object is deleted between fetching two pages.

Nifi Extension Bundles Nifi Jms Bundle Nifi Jms Processors Src Main Resources Docs Org.Apache.Nifi.Jms.Cf.JMSConnectionFactoryProvider AdditionalDetails (`nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/resources/docs/org.apache.nifi.jms.cf.JMSConnectionFactoryProvider/additionalDetails.md`)

JMSConnectionFactoryProvider

Description

This controller service serves as a general factory service to serving vendor specific instances of the
javax.jms.ConnectionFactory. It does so by allowing user to configure vendor specific properties as well as point to
the location of the vendor provided JMS client libraries so the correct implementation of the
javax.jms.ConnectionFactory can be found, loaded, instantiated and served to the dependent processors (see PublishJMS,
ConsumeJMS).

All JMS vendors and ConnectionFactory implementations are supported as long as the configuration values can be set
through set methods (detailed explanation in the last paragraph). However, some helpful accommodation are done for the
following JMS vendors:

Apache ActiveMQ
IBM MQ
TIBCO EMS
Qpid JMS (AMQP 1.0)

This controller service exposes only a single mandatory static configuration property that are required across all
implementations. The rest of the configuration properties are either optional or vendor specific.

The mandatory configuration property is:

JMS Connection Factory Implementation - The fully qualified name of the JMS ConnectionFactory implementation
class. For example:
- Apache
  ActiveMQ - org.apache.activemq.ActiveMQConnectionFactory
- IBM
  MQ - com.ibm.mq.jms.MQQueueConnectionFactory
- TIBCO
  EMS - com.tibco.tibjms.TibjmsQueueConnectionFactory
- Qpid JMS (AMQP
  1.0) - org.apache.qpid.jms.JmsConnectionFactory

The following static configuration properties are optional but required in many cases:

JMS Client Libraries - Path to the directory with additional resources (eg. JARs, configuration files, etc.) to be
added to the classpath (defined as a comma separated list of values). Such resources typically represent target JMS
client libraries for the ConnectionFactory implementation.
JMS Broker URI - URI pointing to the network location of the JMS Message broker. For example:
- Apache ActiveMQ - tcp://myhost:1234 for single broker and failover:(tcp://myhost01:1234,tcp://myhost02:1234)
  for multiple brokers.
- IBM MQ - myhost(1234) for single broker. myhost01(1234),myhost02(1234) for multiple brokers.
- TIBCO EMS - tcp://myhost:1234 for single broker and tcp://myhost01:7222,tcp://myhost02:7222 for multiple
  brokers.
- Qpid JMS (AMQP 1.0) - amqp[s]://myhost:1234 for single broker and failover:(amqp[s]://myhost01:
  1234,amqp[s]://myhost02:1234) for multiple brokers.

The rest of the vendor specific configuration are set through dynamic properties utilizing
the Java Beans convention where a property name is derived
from the set method of the vendor specific ConnectionFactory's implementation. For example,
com.ibm.mq.jms.MQConnectionFactory.setChannel(String) would imply 'channel' property and
com.ibm.mq.jms.MQConnectionFactory.setTransportType(int) would imply 'transportType' property. For the list of
available properties please consult vendor provided documentation. Following is examples of such vendor provided
documentation:

Besides the dynamic properties and set methods described in the previous section, some providers also support
additional configuration via the Broker URI (as query parameters added to the URI):

Qpid JMS (AMQP 1.0)

Sample controller service configuration for IBM MQ

Property	Value	Static/Dynamic	Comments
JMS Connection Factory Implementation	com.ibm.mq.jms.MQQueueConnectionFactory	Static	Vendor provided implementation of QueueConnectionFactory
JMS Client Libraries	/opt/mqm/java/lib	Static	Default installation path of client JAR files on Linux systems
JMS Broker URI	mqhost01(1414),mqhost02(1414)	Static	Connection Name List syntax. Colon separated host/port pair(s) is also supported
JMS SSL Context Service		Static	Only required if using SSL/TLS
channel	TO.BAR	Dynamic	Required when using the client transport mode
queueManager	PQM1	Dynamic	Name of queue manager. Always required.
transportType	1	Dynamic	Constant integer value corresponding to the client transport mode. Default value is "Bindings, then client"

Nifi Extension Bundles Nifi Jms Bundle Nifi Jms Processors Src Main Resources Docs Org.Apache.Nifi.Jms.Cf.JndiJmsConnectionFactoryProvider AdditionalDetails (`nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/resources/docs/org.apache.nifi.jms.cf.JndiJmsConnectionFactoryProvider/additionalDetails.md`)

JMSConnectionFactoryProvider

Capabilities

This Controller Service allows users to reference a JMS Connection Factory that has already been established and made
available via Java Naming and Directory Interface (JNDI) Server. Please see documentation from your JMS Vendor in order
to understand the appropriate values to configure for this service.

A Connection Factory in Java is typically obtained via JNDI in code like below. The comments have been added in to
explain how this maps to the Controller Service's configuration.

java

Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_INITIAL_CONTEXT_FACTORY); // Value for this comes from the "JNDI Initial Context Factory Class" property.
env.put(Context.PROVIDER_URL, JNDI_PROVIDER_URL); // Value for this comes from the "JNDI Provider URL" property.
env.put("My-Environment-Variable","Environment-Variable-Value"); // This is accomplished by added a user-defined property with name "My-Environment-Variable" and value "Environment-Variable-Value"

Context initialContext = new InitialContext(env);
ConnectionFactory connectionFactory = initialContext.lookup(JNDI_CONNECTION_FACTORY_NAME); // Value for this comes from the "JNDI Name of the Connection Factory" property

It is also important to note that, in order for this to work, the class named by the "JNDI Initial Context Factory
Class" must be available on the classpath. The JMS provider specific client classes (like the class of the Connection
Factory object to be retrieved from JNDI) must also be available on the classpath. In NiFi, this is accomplished by
setting the "JNDI / JMS Client Libraries" property to point to one or more .jar files or directories (comma-separated
values).

When the Controller Service is disabled and then re-enabled, it will perform the JNDI lookup again. Once the Connection
Factory has been obtained, though, it will not perform another JNDI lookup until the service is disabled.

Example Configuration

As an example, the following configuration may be used to connect to Active MQ's JMS Broker, using the Connection
Factory provided via their embedded JNDI server:

Property Name	Property Value
JNDI Initial Context Factory Class	org.apache.activemq.jndi.ActiveMQInitialContextFactory
JNDI Provider URL	tcp://jms-broker:61616
JNDI Name of the Connection Factory	ConnectionFactory
JNDI / JMS Client Libraries	/opt/apache-activemq-5.15.2/lib/

The above example assumes that there exists a host that is accessible with hostname "jms-broker" and that is running
Apache ActiveMQ on port 61616 and also that the jar(s) containing the
org.apache.activemq.jndi.ActiveMQInitialContextFactory class and the other JMS client classes can be found within the
/opt/apache-activemq-5.15.2/lib/ directory.

Property Validation

The following component properties include additional validation to restrict allowed values:

JNDI Provider URL

JNDI Provider URL Validation

The default validation for JNDI Provider URL allows the following URL schemes:

file
jgroups
ssl
t3
t3s
tcp
udp
vm

The following Java System property can be configured to override the default allowed URL schemes:

org.apache.nifi.jms.cf.jndi.provider.url.schemes.allowed

The System property must contain a space-separated list of URL schemes. This property can be configured in the
application bootstrap.conf as follows:

java.arg.jndiJmsUrlSchemesAllowed=-Dorg.apache.nifi.jms.cf.jndi.provider.url.schemes.allowed=ssl tcp

Nifi Extension Bundles Nifi Jms Bundle Nifi Jms Processors Src Main Resources Docs Org.Apache.Nifi.Jms.Processors.ConsumeJMS AdditionalDetails (`nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/resources/docs/org.apache.nifi.jms.processors.ConsumeJMS/additionalDetails.md`)

ConsumeJMS

Summary

This processor consumes messages from JMS compliant messaging system and converts them to a FlowFile to be routed to the
next component in the flow.

This processor does two things. It constructs FlowFile by extracting information from the consumed JMS message including
body, standard JMS Headers and Properties. The message body is written to a FlowFile while standard JMS Headers and
Properties are set as FlowFile attributes.

Configuration Details

User Name - [OPTIONAL] User Name used for authentication and authorization when this processor obtains
javax.jms.Connection from the pre-configured javax.jms.ConnectionFactory (see below).
Password - [OPTIONAL] Password used in conjunction with User Name.
Destination Name - [REQUIRED] the name of the javax.jms.Destination. Usually provided by administrator (
e.g., 'topic://myTopic').
Destination Type - [REQUIRED] the type of the javax.jms.Destination. Could be one of 'QUEUE' or 'TOPIC'
Usually provided by the administrator. Defaults to 'QUEUE'.

Connection Factory Configuration

There are multiple ways to configure the Connection Factory for the processor:

Connection Factory Service property - link to a pre-configured controller service (
JndiJmsConnectionFactoryProvider or JMSConnectionFactoryProvider)
JNDI * properties - processor level configuration, the properties are the same as the properties of
JndiJmsConnectionFactoryProvider controller service, the dynamic properties can also be used in this case
JMS * properties - processor level configuration, the properties are the same as the properties of
JMSConnectionFactoryProvider controller service, the dynamic properties can also be used in this case

The preferred way is to use the Connection Factory Service property and a pre-configured controller service. It is also
the most convenient method, because it is enough to configure the controller service once, and then it can be used in
multiple processors.

However, some JMS client libraries may not work with the controller services due to incompatible Java ClassLoader
handling between the 3rd party JMS client library and NiFi. Should you encounter java.lang.ClassCastException errors
when using the controller services, please try to configure the Connection Factory via the 'JNDI *' or the 'JMS *' and
the dynamic properties of the processor. For more details on these properties, see the documentation of the
corresponding controller service (JndiJmsConnectionFactoryProvider for 'JNDI *' and JMSConnectionFactoryProvider
for 'JMS *').

Nifi Extension Bundles Nifi Jms Bundle Nifi Jms Processors Src Main Resources Docs Org.Apache.Nifi.Jms.Processors.PublishJMS AdditionalDetails (`nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/resources/docs/org.apache.nifi.jms.processors.PublishJMS/additionalDetails.md`)

PublishJMS

Summary

This processor publishes the contents of the incoming FlowFile to a JMS compliant messaging system.

This processor does two things. It constructs JMS Message by extracting FlowFile contents (both body and attributes).
Once message is constructed it is sent to a pre-configured JMS Destination. Standard JMS Headers will be extracted from
the FlowFile and set on javax.jms.Message as JMS headers while other FlowFile attributes will be set as properties of
javax.jms.Message. Upon success the incoming FlowFile is transferred to the success Relationship and upon failure
FlowFile is penalized and transferred to the failure Relationship.

Configuration Details

User Name - [OPTIONAL] User Name used for authentication and authorization when this processor obtains
javax.jms.Connection from the pre-configured javax.jms.ConnectionFactory (see below).
Password - [OPTIONAL] Password used in conjunction with User Name.
Destination Name - [REQUIRED] the name of the javax.jms.Destination. Usually provided by administrator
(e.g., 'topic://myTopic').
Destination Type - [REQUIRED] the type of the javax.jms.Destination. Could be one of 'QUEUE' or 'TOPIC'
Usually provided by the administrator. Defaults to 'QUEUE'.

Connection Factory Configuration

There are multiple ways to configure the Connection Factory for the processor:

Connection Factory Service property - link to a pre-configured controller service
(JndiJmsConnectionFactoryProvider or JMSConnectionFactoryProvider)
JNDI * properties - processor level configuration, the properties are the same as the properties of
JndiJmsConnectionFactoryProvider controller service, the dynamic properties can also be used in this case
JMS * properties - processor level configuration, the properties are the same as the properties of
JMSConnectionFactoryProvider controller service, the dynamic properties can also be used in this case

Nifi Extension Bundles Nifi Jolt Bundle Nifi Jolt Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Jolt.JoltTransformJSON AdditionalDetails (`nifi-extension-bundles/nifi-jolt-bundle/nifi-jolt-processors/src/main/resources/docs/org.apache.nifi.processors.jolt.JoltTransformJSON/additionalDetails.md`)

JoltTransformJSON

Usage Information

The Jolt utilities processing JSON are not stream based therefore large JSON document transformation may consume large
amounts of memory. Currently, UTF-8 FlowFile content and Jolt specifications are supported. A specification can be
defined using Expression Language where attributes can be referred either on the left or right hand side within the
specification syntax. Custom Jolt Transformations (that implement the Transform interface) are supported. Modules
containing custom libraries which do not existing on the current class path can be included via the custom module
directory property. Note: When configuring a processor if user selects of the Default transformation yet provides a
Chain specification the system does not alert that the specification is invalid and will produce failed flow files. This
is a known issue identified within the Jolt library.

Nifi Extension Bundles Nifi Kafka Bundle Nifi Kafka Processors Src Main Resources Docs Org.Apache.Nifi.Kafka.Processors.ConsumeKafka AdditionalDetails (`nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md`)

ConsumeKafka

Output Strategies

This processor offers multiple output strategies (configured via processor property 'Output Strategy') for converting Kafka records into FlowFiles.

'Use Content as Value' (the default) emits FlowFile records containing only the Kafka record value.
'Use Wrapper' emits FlowFile records containing the Kafka record key, value, and headers, as well as additional metadata from the Kafka record.
'Inject Metadata' emits FlowFile records containing the Kafka record value into which a sub record field has been added to hold metadata, headers and key information.

The record schema that is used when "Use Wrapper" is selected is as follows (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [{
      "name": "key",
      "type": [{
          < Schema is determined by the Key Record Reader, or will be "string" or "bytes", depending on the "Key Format" property (see below for more details) >
        }, "null"]
    },
    {
      "name": "value",
      "type": [
        {
          < Schema is determined by the Record Reader >
        },
        "null"
      ]
    },
    {
      "name": "headers",
      "type": [
        { "type": "map", "values": "string", "default": {}},
        "null"]
    },
    {
      "name": "metadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "topic", "type": ["string", "null"] },
            { "name": "partition", "type": ["int", "null"] },
            { "name": "offset", "type": ["int", "null"] },
            { "name": "timestamp", "type": ["long", "null"] }
          ]
        },
        "null"
      ]
    }
  ]
}

The record schema that is used when "Inject Metadata" is selected is as follows (in Avro format):

json

{
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
      < Fields as determined by the Record Reader for the Kafka message >,
    {
      "name": "kafkaMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "topic", "type": ["string", "null"] },
            { "name": "partition", "type": ["int", "null"] },
            { "name": "offset", "type": ["int", "null"] },
            { "name": "timestamp", "type": ["long", "null"] },
            {
              "name": "headers",
              "type": [ { "type": "map", "values": "string", "default": {}}, "null"]
            },
            {
              "name": "key",
              "type": [{
                < Schema is determined by the Key Record Reader, or will be "string" or "bytes", depending on the "Key Format" property (see below for more details) >
              }, "null"]
            }
          ]
        },
        "null"
      ]
    }
  ]
}

If the Output Strategy property is set to 'Use Wrapper' or 'Inject Metadata', an additional processor configuration property ('Key Format') is activated. This property is used to specify how the Kafka Record's key should be written out to the FlowFile. The possible values for 'Key Format' are as follows:

'Byte Array' supplies the Kafka Record Key as a byte array, exactly as they are received in the Kafka record.
'String' converts the Kafka Record Key bytes into a string using the UTF-8 character encoding. (Failure to parse the key bytes as UTF-8 will result in the record being routed to the 'parse.failure' relationship.)
'Record' converts the Kafka Record Key bytes into a deserialized NiFi record, using the associated 'Key Record Reader' controller service. If the Key Format property is set to 'Record', an additional processor configuration property name 'Key Record Reader' is made available. This property is used to specify the Record Reader to use in order to parse the Kafka Record's key as a Record.

Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy "Use Wrapper" is selected:

json

[
  {
    "key": {
      "name": "Acme",
      "number": "AC1234"
    },
    "value": {
      "address": "1234 First Street",
      "zip": "12345",
      "account": {
        "name": "Acme",
        "number": "AC1234"
      }
    },
    "headers": {
      "attributeA": "valueA",
      "attributeB": "valueB"
    },
    "metadata": {
      "topic": "accounts",
      "partition": 0,
      "offset": 0,
      "timestamp": 0
    }
  }
]

Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy "Inject Metadata" is selected:

json

[
  {
    "address": "1234 First Street",
    "zip": "12345",
    "account": {
      "name": "Acme",
      "number": "AC1234"
    },
    "kafkaMetadata": {
      "topic": "accounts",
      "partition": 0,
      "offset": 0,
      "timestamp": 0,
      "headers": {
        "attributeA": "valueA",
        "attributeB": "valueB"
      },
      "key": {
        "name": "Acme",
        "number": "AC1234"
      }
    }
  }
]

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.DeleteMongo AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.DeleteMongo/additionalDetails.md`)

DeleteMongo

This processor deletes from Mongo using a user-provided query that is provided in the body of a flowfile. It must be a
valid JSON document. The user has the option of deleting a single document or all documents that match the criteria.
That behavior can be configured using the related configuration property. In addition, the processor can be configured
to regard a failure to delete any documents as an error event, which would send the flowfile with the query to the
failure relationship.

Example Query

json

{
  "username": "john.smith",
  "recipient": "jane.doe"
}

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.GetMongo AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.GetMongo/additionalDetails.md`)

GetMongo

This processor runs queries against a MongoDB instance or cluster and writes the results to a flowfile. It allows input,
but can run standalone as well.

Specifying the Query

The query can be specified in one of three ways:

Query configuration property.
Query Attribute configuration property.
FlowFile content.

If a value is specified in either of the configuration properties, it will not look in the FlowFile content for a query.

Limiting/Shaping Results

The following options for limiting/shaping results are available:

Limit - limit the number of results. This should not be confused with the "batch size" option which is a setting for
the underlying MongoDB driver to tell it how many items to retrieve in each poll of the server.
Sort - sort the result set. Requires a JSON document like { "someDate": -1 }
Projection - control which fields to return. Exampe, which would remove _id: { "_id": 0 }

Misc Options

Results Per FlowFile, if set, creates a JSON array out of a batch of results and writes the result to the output. Pretty
Print, if enabled, will format the JSON data to be easy read by a human (ex. proper indentation of fields).

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.GetMongoRecord AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.GetMongoRecord/additionalDetails.md`)

GetMongoRecord

This processor runs queries against a MongoDB instance or cluster and writes the results to a flowfile. It allows input,
but can run standalone as well. It is a record-aware version of the GetMongo processor.

Specifying the Query

The query can be specified in one of three ways:

Query configuration property.
Query Attribute configuration property.
FlowFile content.

If a value is specified in either of the configuration properties, it will not look in the FlowFile content for a query.

Limiting/Shaping Results

The following options for limiting/shaping results are available:

Limit - limit the number of results. This should not be confused with the "batch size" option which is a setting for
the underlying MongoDB driver to tell it how many items to retrieve in each poll of the server.
Sort - sort the result set. Requires a JSON document like { "someDate": -1 }
Projection - control which fields to return. Exampe, which would remove _id: { "_id": 0 }

Misc Options

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.PutMongo AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.PutMongo/additionalDetails.md`)

PutMongo

This processor is a general purpose processor for inserting, upserting and updating MongoDB collections.

Inserting Documents

Each flowfile is assumed to contain only a single MongoDB document to be inserted. The contents must be valid JSON. The
input the Mongo shell accepts should not be confused with valid JSON. It does not support batch writes at this time.

Updating and Upserting

Update Modes

There are two methods for choosing what gets written to a document when updating:

Whole document - the entire document is replaced with the contents of the flowfile.
With Operators Enabled - the document in the flowfile content will be assumed to have update operators such as $set
and will be used to update particular fields. The whole document will not be replaced.

There are two ways to update:

Update Key - use one or more keys from the document.
Update Query - use a totally separate query that is not derived from the document.

Update Key

The update key method takes keys from the document and builds a query from them. It will attempt to parse the _id
field as an ObjectID type if that is one of the keys that is specified in the configuration field. Multiple keys can
be specified by separating them with commas. This configuration field supports Expression Language, so it can be derived
in part or entirely from flowfile attributes.

Update Query

The update query method takes a valid JSON document as its value and uses it to find one or more documents to update.
This field supports Expression Language, so it can be derived in part or entirely from flowfile attributes. It is
possible, for instance, to put an attribute named update_query on a flowfile and specify ${update_query} in the
configuration field, so it will load the value from the flowfile.

Upserts

If the upsert mode is enabled, PutMongo will insert a new document that matches the search criteria (be it a
user-supplied query or one built from update keys) and give it the properties that are specified in the JSON document
provided in the flowfile content. This feature should be used carefully, as it can result in incomplete data being added
to MongoDB.

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.PutMongoBulkOperation AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.PutMongoBulkOperation/additionalDetails.md`)

PutMongoBulkOperation

This processor runs bulk updates against MongoDB collections. The flowfile content is expected to be a JSON array with
bulk write operations as described in
the manual for db.collection.bulkWrite.

You can use all (currently 6) operators described there. The flowfile content is returned as-is. You can merge many
operations into one - and get massive performance improvements.

Example:

The following is an example flowfile content that does two things: insert a new document, and update all documents where
value of hey is greater than zero.

json

[
  {
    "insertOne": {
      "document": {
        "ho": 42
      }
    }
  },
  {
    "updateMany": {
      "filter": {
        "hey": {
          "$gt": 0
        }
      },
      "update": {
        "$inc": {
          "hey": 2
        }
      }
    }
  }
]

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.RunMongoAggregation AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.RunMongoAggregation/additionalDetails.md`)

RunMongoAggregation

This processor runs a MongoDB aggregation query based on user-defined settings. The following is an example of such a
query (and what the expected input looks like):

json

[
  {
    "$project": {
      "domain": 1
    },
    "$group": {
      "_id": {
        "domain": "$domain"
      },
      "total": {
        "$sum": 1
      }
    }
  }
]

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.Gridfs.DeleteGridFS AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.gridfs.DeleteGridFS/additionalDetails.md`)

DeleteGridFS

This processor retrieves one or more files from GridFS. The query to execute can be either provided in the query
configuration parameter or generated from the value pulled from the filename configuration parameter. Upon successful
execution, it will append the query that was executed as an attribute on the flowfile that was processed.

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.Gridfs.FetchGridFS AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.gridfs.FetchGridFS/additionalDetails.md`)

FetchGridFS

This processor retrieves one or more files from GridFS. The query can be provided in one of three ways:

Query configuration parameter.
Built for you by configuring the filename parameter. (Note: this is just a filename, Mongo queries cannot be embedded
in the field).
Retrieving the query from the flowfile contents.

The processor can also be configured to either commit only once at the end of a fetch operation or after each file that
is retrieved. Multiple commits is generally only necessary when retrieving a lot of data from GridFS as measured in
total data size, not file count, to ensure that the disks NiFi is using are not overloaded.

Nifi Extension Bundles Nifi Mongodb Bundle Nifi Mongodb Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mongodb.Gridfs.PutGridFS AdditionalDetails (`nifi-extension-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/resources/docs/org.apache.nifi.processors.mongodb.gridfs.PutGridFS/additionalDetails.md`)

PutGridFS

This processor puts a file with one or more user-defined metadata values into GridFS in the configured bucket. It allows
the user to define how big each file chunk will be during ingestion and provides some ability to intelligently attempt
to enforce file uniqueness using filename or hash values instead of just relying on a database index.

GridFS File Attributes

PutGridFS allows for flowfile attributes that start with a configured prefix to be added to the GridFS document. These
can be very useful later when working with GridFS for providing metadata about a file.

Chunk Size

GridFS splits up file into chunks within Mongo documents as the file is ingested into the database. The chunk size
configuration parameter configures the maximum size of each chunk. This field should be left at its default value unless
there is a specific business case to increase or decrease it.

Uniqueness Enforcement

There are four operating modes:

No enforcement at the application level.
Enforce by unique file name.
Enforce by unique hash value.
Use both hash and file name.

The hash value by default is taken from the attribute hash.value which can be generated by configuring a HashContent
processor upstream of PutGridFS. Both this and the name option use a query on the existing data to see if a file
matching that criteria exists before attempting to write the flowfile contents.

Nifi Extension Bundles Nifi Mqtt Bundle Nifi Mqtt Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Mqtt.ConsumeMQTT AdditionalDetails (`nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/resources/docs/org.apache.nifi.processors.mqtt.ConsumeMQTT/additionalDetails.md`)

ConsumeMQTT

The MQTT messages are always being sent to subscribers on a topic regardless of how frequently the processor is
scheduled to run. If the 'Run Schedule' is significantly behind the rate at which the messages are arriving to this
processor, then a back-up can occur in the internal queue of this processor. Each time the processor is scheduled, the
messages in the internal queue will be written to FlowFiles. In case the internal queue is full, the MQTT client will
try for up to 1 second to add the message into the internal queue. If the internal queue is still full after this time,
an exception saying that 'The subscriber queue is full' would be thrown, the message would be dropped and the client
would be disconnected. In case the QoS property is set to 0, the message would be lost. In case the QoS property is set
to 1 or 2, the message will be received after the client reconnects.

Nifi Extension Bundles Nifi Network Bundle Nifi Network Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Network.ParseNetflowv5 AdditionalDetails (`nifi-extension-bundles/nifi-network-bundle/nifi-network-processors/src/main/resources/docs/org.apache.nifi.processors.network.ParseNetflowv5/additionalDetails.md`)

Netflowv5Parser

Netflowv5Parser processor parses the ingress netflowv5 datagram format and transfers it either as flowfile attributes or JSON object. Netflowv5 format has predefined schema named "template" for parsing the netflowv5 record. More information: RFC-netflowv5

Netflowv5 JSON Output Schema

json

{
  "port": int,
  "format": string,
  "header": {
    "version": int,
    "count": int,
    "sys_uptime": long,
    "unix_secs": long,
    "unix_nsecs": long,
    "flow_sequence": long,
    "engine_type": short,
    "engine_id": short,
    "sampling_interval": int
  },
  "record": {
    "srcaddr": string,
    "dstaddr": string,
    "nexthop": string,
    "input": int,
    "output": int,
    "dPkts": long,
    "dOctets": long,
    "first": long,
    "last": long,
    "srcport": int,
    "dstport": int,
    "pad1": short,
    "tcp_flags": short,
    "prot": short,
    "tos": short,
    "src_as": int,
    "dst_as": int,
    "src_mask": short,
    "dst_mask": short,
    "pad2": int
  }
}

Nifi Extension Bundles Nifi Poi Bundle Nifi Poi Services Src Main Resources Docs Org.Apache.Nifi.Excel.ExcelReader AdditionalDetails (`nifi-extension-bundles/nifi-poi-bundle/nifi-poi-services/src/main/resources/docs/org.apache.nifi.excel.ExcelReader/additionalDetails.md`)

ExcelReader

The ExcelReader allows for interpreting input data as delimited Records. Each row in an Excel spreadsheet is a record
and each cell is considered a field. The reader allows for choosing which row to start from and which sheets in a
spreadsheet to ingest. When using the "Use Starting Row" strategy, the field names will be assumed to be the column
names from the configured starting row. If there are any column(s) from the starting row which are blank, they are
automatically assigned a field name using the cell number prefixed with "column_". When using the "Infer Schema"
strategy, the field names will be assumed to be the cell numbers of each column prefixed with "column_". Otherwise, the
names of fields can be supplied when specifying the schema by using the Schema Text or looking up the schema in a Schema
Registry.

Schemas and Type Coercion

When a record is parsed from incoming data, it is separated into fields. Each of these fields is then looked up against
the configured schema (by field name) in order to determine what the type of the data should be. If the field is not
present in the schema, that field is omitted from the Record. If the field is found in the schema, the data type of the
received data is compared against the data type specified in the schema. If the types match, the value of that field is
used as-is. If the schema indicates that the field should be of a different type, then the Controller Service will
attempt to coerce the data into the type specified by the schema. If the field cannot be coerced into the specified
type, an Exception will be thrown.

The following rules apply when attempting to coerce a field value from one data type to another:

Excel stores all numeric types as a Double which can be coerced to any other numeric data type (Byte, Short, Int, Long, Float).
Any data type can be coerced into a String type. Please note since Excel stores all numbers as a Double, a large number coerced to a String will result in a string representation of the number in scientific notation. If this is not desired, then coerce the number to a Long numeric type.
Any numeric value can be coerced into a Date, Time, or Timestamp type, by assuming that the Long value is the number
of milliseconds since epoch (Midnight GMT, January 1, 1970).
A String value can be coerced into a Date, Time, or Timestamp type, if its format matches the configured "Date
Format," "Time Format," or "Timestamp Format."
A String value can be coerced into a numeric value if the value is of the appropriate type. For example, the String
value 8 can be coerced into any numeric type. However, the String value 8.2 can be coerced into a Double or Float
type but not an Integer.
A String value of "true" or "false" (regardless of case) can be coerced into a Boolean value.
A String value that is not empty can be coerced into a Char type. If the String contains more than 1 character, the
first character is used and the rest of the characters are ignored.
Any "date/time" type (Date, Time, Timestamp) can be coerced into any other "date/time" type.
Any "date/time" type can be coerced into a Long type, representing the number of milliseconds since epoch (Midnight
GMT, January 1, 1970).
Any "date/time" type can be coerced into a String. The format of the String is whatever DateFormat is configured for
the corresponding property (Date Format, Time Format, Timestamp Format property).

If none of the above rules apply when attempting to coerce a value from one data type to another, the coercion will fail
and an Exception will be thrown.

Use Starting Row and Schema Inference

While NiFi's Record API does require that each Record have a schema, it is often convenient to infer the schema based on
the values in the data, rather than having to manually create a schema. This is accomplished by selecting either value
of "Use Starting Row" or "Infer Schema" for the "Schema Access Strategy" property. When using the "Use Starting Row"
strategy, the Reader will determine the schema by parsing the first ten rows after the configured starting row of the
data in the FlowFile all the while keeping track of all fields that it has encountered and the type of each field. A
schema is then formed that encompasses all encountered fields. A schema can even be inferred if there are blank lines
within those ten rows, but if they are all blank, then this strategy will fail to create a schema. When using the "Infer
Schema" strategy, the Reader will determine the schema by first parsing all data in the FlowFile, keeping track of all
fields that it has encountered and the type of each field. Once all data has been parsed, a schema is formed that
encompasses all fields that have been encountered.

When using the "Use Starting Row" strategy, a row having more cells than the header row causes schema inference to
fail. This often happens with spreadsheets edited by hand, where cells to the right of the data are blank but formatted.
Setting the "Ignore Cells Beyond Header" property to true ignores the cells beyond the last header column instead. Blank
cells are ignored silently, while a warning is logged for each sheet where the ignored cells contain a value.

A common concern when inferring schemas is how to handle the condition of two values that have different types. For
example, consider a FlowFile with the following two records:

text

name, age
John, 8
Jane, Ten

It is clear that the "name" field will be inferred as a STRING type. However, how should we handle the "age" field?
Should the field be an CHOICE between INT and STRING? Should we prefer LONG over INT? Should we just use a STRING?
Should the field be considered nullable?

To help understand how this Record Reader infers schemas, we have the following list of rules that are followed in the
inference logic:

All fields are inferred to be nullable.
When two values are encountered for the same field in two different records (or two values are encountered for an
ARRAY type), the inference engine prefers to use a "wider" data type over using a CHOICE data type. A data type "A" is
said to be wider than data type "B" if and only if data type "A" encompasses all values of "B" in addition to other
values. For example, the LONG type is wider than the INT type but not wider than the BOOLEAN type (and BOOLEAN is also
not wider than LONG). INT is wider than SHORT. The STRING type is considered wider than all other types except MAP,
RECORD, ARRAY, and CHOICE.
Before inferring the type of value, leading and trailing whitespace are removed. Additionally, if the value is
surrounded by double-quotes ("), the double-quotes are removed. Therefore, the value 16 is interpreted the same as
"16". Both will be interpreted as an INT. However, the value " 16" will be inferred as a STRING type because the
white space is enclosed within double-quotes, which means that the white space is considered part of the value.
If the "Time Format," "Timestamp Format," or "Date Format" properties are configured, any value that would otherwise
be considered a STRING type is first checked against the configured formats to see if it matches any of them. If the
value matches the Timestamp Format, the value is considered a Timestamp field. If it matches the Date Format, it is
considered a Date field. If it matches the Time Format, it is considered a Time field. In the unlikely event that the
value matches more than one of the configured formats, they will be matched in the order: Timestamp, Date, Time. I.e.,
if a value matched both the Timestamp Format and the Date Format, the type that is inferred will be Timestamp. Because
parsing dates and times can be expensive, it is advisable not to configure these formats if dates, times, and
timestamps are not expected, or if processing the data as a STRING is acceptable. For use cases when this is
important, though, the inference engine is intelligent enough to optimize the parsing by first checking several very
cheap conditions. For example, the string's length is examined to see if it is too long or too short to match the
pattern. This results in far more efficient processing than would result if attempting to parse each string value as a
timestamp.
The MAP type is never inferred.
The ARRAY type is never inferred.
The RECORD type is never inferred.
If a field exists but all values are null, then the field is inferred to be of type STRING.

Caching of Inferred Schemas

This Record Reader requires that if a schema is to be inferred, that all records be read in order to ensure that the
schema that gets inferred is applicable for all records in the FlowFile. However, this can become expensive, especially
if the data undergoes many different transformations. To alleviate the cost of inferring schemas, the Record Reader can
be configured with a "Schema Inference Cache" by populating the property with that name. This is a Controller Service
that can be shared by Record Readers and Record Writers.

Whenever a Record Writer is used to write data, if it is configured with a "Schema Cache," it will also add the schema
to the Schema Cache. This will result in an identifier for that schema being added as an attribute to the FlowFile.

Whenever a Record Reader is used to read data, if it is configured with a "Schema Inference Cache", it will first look
for a "schema.cache.identifier" attribute on the FlowFile. If the attribute exists, it will use the value of that
attribute to lookup the schema in the schema cache. If it is able to find a schema in the cache with that identifier,
then it will use that schema instead of reading, parsing, and analyzing the data to infer the schema. If the attribute
is not available on the FlowFile, or if the attribute is available but the cache does not have a schema with that
identifier, then the Record Reader will proceed to infer the schema as described above.

The end result is that users are able to chain together many different Processors to operate on Record-oriented data.
Typically, only the first such Processor in the chain will incur the "penalty" of inferring the schema. For all other
Processors in the chain, the Record Reader is able to simply lookup the schema in the Schema Cache by identifier. This
allows the Record Reader to infer a schema accurately, since it is inferred based on all data in the FlowFile, and still
allows this to happen efficiently since the schema will typically only be inferred once, regardless of how many
Processors handle the data.

Examples

Example 1

As an example, consider a FlowFile whose contents are an Excel spreadsheet whose only sheet consists of the following:

text

id, name, balance, join_date, notes
1, John, 48.23, 04/03/2007, "Our very   first customer!"
2, Jane, 1245.89, 08/22/2009,	  
3, Frank Franklin, "48481.29", 04/04/2016,

Additionally, let's consider that this Controller Service is configured to skip the first line and is configured with
the Schema Registry pointing to an AvroSchemaRegistry which contains the following schema:

json

{
  "namespace": "nifi",
  "name": "balances",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "balance",
      "type": "double"
    },
    {
      "name": "join_date",
      "type": {
        "type": "int",
        "logicalType": "date"
      }
    },
    {
      "name": "notes",
      "type": "string"
    }
  ]
}

In the example above, we see that the 'join_date' column is a Date type. In order for the Excel Reader to be able to
properly parse a value as a date, we need to provide the reader with the date format to use. In this example, we would
configure the Date Format property to be MM/dd/yyyy to indicate that it is a two-digit month, followed by a two-digit
day, followed by a four-digit year - each separated by a slash. In this case, the result will be that this FlowFile
consists of 3 different records. The first record will contain the following values:

Field Name	Field Value
id	1
name	John
balance	48.23
join_date	04/03/2007
notes	Our very first customer!

The second record will contain the following values:

Field Name	Field Value
id	2
name	Jane
balance	1245.89
join_date	08/22/2009
notes

The third record will contain the following values:

Field Name	Field Value
id	3
name	Frank Franklin
balance	48481.29
join_date	04/04/2016
notes

Nifi Extension Bundles Nifi Protobuf Bundle Nifi Protobuf Services Src Main Resources Docs Org.Apache.Nifi.Services.Protobuf.ProtobufReader AdditionalDetails (`nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/docs/org.apache.nifi.services.protobuf.ProtobufReader/additionalDetails.md`)

ProtobufReader

The ProtobufReader Controller Service reads and parses a Protocol Buffers Message from binary format and creates a
Record object. The Controller Service must be configured with the same '.proto' file that was used for the Message
encoding, and the fully qualified name of the Message type including its package (e.g. mypackage.MyMessage). The Reader
will always generate one record from the input data which represents the provided Protocol Buffers Message type. Further
information about Protocol Buffers can be found here: protobuf.dev

Data Type Mapping

When a record is parsed from incoming data, the Controller Service is going to map the Proto Message field types to the
corresponding NiFi data types. The mapping between the provided Message fields and the encoded input is always based on
the field tag numbers. When a field is defined as 'repeated' then it's data type will be an array with data type of it's
originally specified type. The following tables show which proto field type will correspond to which NiFi field type
after the conversion.

Scalar Value Types

Proto Type	Proto Wire Type	NiFi Data Type
double	fixed64	double
float	fixed32	float
int32	varint	int
int64	varint	long
uint32	varint	long
uint64	varint	bigint
sint32	varint	long
sint64	varint	long
fixed32	fixed32	long
fixed64	fixed64	bigint
sfixed32	varint	int
sfixed64	varint	long
bool	varint	boolean
string	length delimited	string
bytes	length delimited	array[byte]

Composite Value Types

Proto Type	Proto Wire Type	NiFi Data Type
message	length delimited	record
enum	varint	enum
map	length delimited	map
oneof	-	choice

Schemas and Type Coercion

When a record is parsed from incoming data, it is separated into fields. Each of these fields is then looked up against
the configured schema (by field name) in order to determine what the type of the data should be. If the field is not
present in the schema, that field will be stored in the Record's value list on its original type. If the field is found
in the schema, the data type of the received data is compared against the data type specified in the schema. If the
types match, the value of that field is used as-is. If the schema indicates that the field should be of a different
type, then the Controller Service will attempt to coerce the data into the type specified by the schema. If the field
cannot be coerced into the specified type, an Exception will be thrown.

The following rules apply when attempting to coerce a field value from one data type to another:

Any data type can be coerced into a String type.
Any numeric data type (Int, Long, Float, Double) can be coerced into any other numeric data type.
Any numeric value can be coerced into a Date, Time, or Timestamp type, by assuming that the Long value is the number
of milliseconds since epoch (Midnight GMT, January 1, 1970).
A String value can be coerced into a Date, Time, or Timestamp type, if its format matches the configured "Date
Format," "Time Format," or "Timestamp Format."
A String value can be coerced into a numeric value if the value is of the appropriate type. For example, the String
value 8 can be coerced into any numeric type. However, the String value 8.2 can be coerced into a Double or Float
type but not an Integer.
A String value of "true" or "false" (regardless of case) can be coerced into a Boolean value.
A String value that is not empty can be coerced into a Char type. If the String contains more than 1 character, the
first character is used and the rest of the characters are ignored.

If none of the above rules apply when attempting to coerce a value from one data type to another, the coercion will fail
and an Exception will be thrown.

Schema Access Strategy

Beside the common Schema Access strategies like getting the schema from property value or accessing it from Schema
Registry, the ProtobufReader Controller Service offers another access strategy option called "Generate from Proto file".
When using this strategy, the Reader will generate the Record Schema from the provided '.proto' file and Message type.
This is a recommended strategy when the user doesn't want to manually create the schema or when no type coercion is
needed.

Protobuf Any Field Type

Protocol Buffers offers further Message types called Well-Known Types. These are additionally provided messages that
defines complex structured types and wrappers for scalar types. The Any type is one of these Well-Known Types which is
used to store an arbitrary serialized Message along with a URL that describes the type of the serialized Message. Since
the Message type and the embedded Message will be available only when the Any Message is already populated with data,
the ProtobufReader needs to do this Message processing at data conversion time. The Reader is capable to generate schema
for the embedded Message in the Any field and replace it in the result Record schema.

Example

There is a Message called 'TestMessage' which has only one field that is an Any typed field. There is another Message
called 'NestedMessage' that we would like to add as serialized Message in the value of 'anyField'.

text

message Any {
    string type_url = 1;
    bytes value = 2; 
}

message TestMessage {
    google.protobuf.Any anyField = 3; 
}

message NestedMessage {
    string field_1 = 1;
    string field_2 = 2;     
    string field_3 = 3;
}

With normal data conversion our result would look like this:

json

{
  anyField: {
    type_url: "type.googleapis.com/NestedMessage"
    value: [
      84,
      101,
      115,
      116,
      32,
      98,
      121,
      116,
      101,
      115
    ]
  }
}

Result after the Protobuf Reader replaces the Any Message's fields with the processed embedded Message:

json

{
  anyField: {
    field_1: "value 1",
    field_2: "value 2",
    field_3: "value 3"
  }
}

Nifi Extension Bundles Nifi Salesforce Bundle Nifi Salesforce Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Salesforce.PutSalesforceObject AdditionalDetails (`nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.PutSalesforceObject/additionalDetails.md`)

PutSalesforceObject

Description

Objects in Salesforce are database tables, their rows are known as records, and their columns are called fields. The
PutSalesforceObject creates a new a Salesforce record in a Salesforce object. The Salesforce object must be set as the
"objectType" attribute of an incoming flowfile.
Check Salesforce documentation
for object types and metadata. The processor utilizes NiFi record-based processing to allow arbitrary input format.

Example

If the "objectType" is set to "Account", the following JSON input will create two records in the Account object with the
names "SampleAccount1" and "SampleAccount2".

json

[
  {
    "name": "SampleAccount1",
    "phone": "1111111111",
    "website": "www.salesforce1.com",
    "numberOfEmployees": "100",
    "industry": "Banking"
  },
  {
    "name": "SampleAccount2",
    "phone": "22222222",
    "website": "www.salesforce2.com",
    "numberOfEmployees": "200",
    "industry": "Banking"
  }
]

Nifi Extension Bundles Nifi Salesforce Bundle Nifi Salesforce Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Salesforce.QuerySalesforceObject AdditionalDetails (`nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.QuerySalesforceObject/additionalDetails.md`)

QuerySalesforceObject

Description

Objects in Salesforce are database tables, their rows are known as records, and their columns are called fields. The
QuerySalesforceObject processor queries Salesforce objects and retrieves their records. The processor constructs the
query from processor properties or executes a custom SOQL (Salesforce Object Query Language) query and retrieves the
result record dataset using the Salesforce REST API. The 'Query Type' processor property allows the query to be built in
two ways. The 'Property Based Query' option allows to define a 'SELECT from ' type query,
with the fields defined in the 'Field Names' property and the Salesforce object defined in the 'sObject Name' property,
whereas the 'Custom Query' option allows you to supply an arbitrary SOQL query. By using 'Custom Query', the processor
can accept an optional input flowfile and reference the flowfile attributes in the query. However, incremental loading
and record-based processing are only supported in 'Property Based Queries'.

OAuth2 Access Token Provider Service

The OAuth2 Access Token Provider Service handles Salesforce REST API authorization. In order to use OAuth2
authorization, create a new StandardOauth2AccessTokenProvider service and configure it as follows.

Authorization Server URL: It is the concatenation of the Salesforce URL and the token request service URL (
/services/oauth2/token).
Grant Type: User Password.
Username: The email address registered in the Salesforce account.
Password: For the Password a Security token must be requested. Go to Profile -> Settings and under the Reset My
Security Token option, request one, which will be sent to the registered email address. The password is made up of the
Salesforce account password and the Security token concatenated together without a space.
Client ID: Create a new Connected App within Salesforce. Go to Setup -> On the left search panel find App Manager ->
Create New Connected App. Once it’s done, the Consumer Key goes to the Client ID property.
Client Secret: Available on the Connected App page under Consumer Secret.

Age properties

The age properties are important to avoid processing duplicate records. Age filtering provides a sliding window that
starts with the processor’s prior run time and ends with the current run time minus the age delay. Only records that are
within the sliding window are queried and processed. On the processor, the Age Field property must be a datetime field
of the queried object, this will be subject to the condition that it is greater than the processor's previous but less
than the current run time (e.g. LastModifiedDate). The first run, for example, will query records whose LastModifiedDate
field is earlier than the current run time. The second will look for records with LastModifiedDate fields that are later
than the previous run time but earlier than the current run time.

The processor uses the Initial Age Filter as a specific timestamp that sets the beginning of the sliding window from
which processing builds the initial query. The format must adhere to the Salesforce SOQL standards (see Salesforce
documentation). The Age Delay moves the time of the records to be processed earlier than the current run time if
necessary.

The 'sObject Name' and 'Field Names' properties can reference attributes of an incoming flowfile. When the Age Field
property is set, the processor stores the end of the last time window separately for each queried sObject, so the same
processor can incrementally query several objects. The Age Field must be a datetime field that exists on every object
that is queried, as the name is shared across them.

Nifi Extension Bundles Nifi Scripting Bundle Nifi Scripting Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Script.ExecuteScript AdditionalDetails (`nifi-extension-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.md`)

ExecuteScript

Description

The ExecuteScript Processor provides the ability to use a scripting language in order to leverage the NiFi API to
perform tasks such as the following:

Read content and/or attributes from an incoming FlowFile
Create a new FlowFile (with or without a parent)
Write content and/or attributes to an outgoing FlowFile
Interact with the ProcessSession to transfer FlowFiles to relationships
Read/write to the State Manager to keep track of variables across executions of the processor

Notes:

ExecuteScript uses the JSR-223 Script Engine API to evaluate scripts, so the use of idiomatic language structure is
sometimes limited. For example, in the case of Groovy, there is a separate ExecuteGroovyScript processor that allows
you to do many more idiomatic Groovy tasks. For example, it's easier to interact with Controller Services via
ExecuteGroovyScript vs. ExecuteScript (see the ExecuteGroovyScript documentation for more details)

Variable Bindings

The Processor expects a user defined script that is evaluated when the processor is triggered. The following variables
are available to the scripts:

Variable Name	Description	Variable Class
session	This is a reference to the ProcessSession assigned to the processor. The session allows you to perform operations on FlowFiles such as create(), putAttribute(), and transfer(), as well as read() and write()	ProcessSession
context	This is a reference to the ProcessContext for the processor. It can be used to retrieve processor properties, relationships, Controller Services, and the State Manager.	ProcessContext
log	This is a reference to the ComponentLog for the processor. Use it to log messages to NiFi, such as log.info('Hello world!')	ComponentLog
REL_SUCCESS	This is a reference to the "success" relationship defined for the processor. It could also be inherited by referencing the static member of the parent class (ExecuteScript), this is a convenience variable. It also saves having to use the fully-qualified name for the relationship.	Relationship
REL_FAILURE	This is a reference to the "failure" relationship defined for the processor. As with REL_SUCCESS, it could also be inherited by referencing the static member of the parent class (ExecuteScript), this is a convenience variable. It also saves having to use the fully-qualified name for the relationship.	Relationship
Dynamic Properties	Any dynamic (user-defined) properties defined in ExecuteScript are passed to the script engine as variables set to the PropertyValue object corresponding to the dynamic property. This allows you to get the String value of the property, but also to evaluate the property with respect to NiFi Expression Language, cast the value as an appropriate data type (e.g., Boolean), etc. Because the dynamic property name becomes the variable name for the script, you must be aware of the variable naming properties for the chosen script engine. For example, Groovy does not allow periods (.) in variable names, so an error will occur if "my.property" was a dynamic property name. Interaction with these variables is done via the NiFi Java API. The 'Dynamic Properties' section below will discuss the relevant API calls as they are introduced.	PropertyValue

Example Scripts

Get an incoming FlowFile from the session

Use Case: You have incoming connection(s) to ExecuteScript and want to retrieve one FlowFile from the queue(s) for
processing.

Approach: Use the get() method from the session object. This method returns the FlowFile that is next highest
priority FlowFile to process. If there is no FlowFile to process, the method will return null. NOTE: It is possible to
have null returned even if there is a steady flow of FlowFiles into the processor. This can happen if there are multiple
concurrent tasks for the processor, and the other task(s) have already retrieved the FlowFiles. If the script requires a
FlowFile to continue processing, then it should immediately return if null is returned from session.get()

Groovy

groovy

flowFile = session.get()
if (!flowFile) return

Get multiple incoming FlowFiles from the session:

Use Case: You have incoming connection(s) to ExecuteScript and want to retrieve multiple FlowFiles from the queue(s)
for processing.

Approach: Use the get(maxResults) method from the session object. This method returns up to maxResults FlowFiles
from the work queue. If no FlowFiles are available, an empty list is returned (the method does not return null). NOTE:
If multiple incoming queues are present, the behavior is unspecified in terms of whether all queues or only a single
queue will be polled in a single call. Having said that, the observed behavior (for both NiFi 1.1.0+ and before) is
described here.

Examples:

Groovy

groovy

flowFileList = session.get(100)
if (!flowFileList.isEmpty()) {
    flowFileList.each { flowFile ->
// Process each FlowFile here
    }
}

Create a new FlowFile

Use Case: You want to generate a new FlowFile to send to the next processor.

Approach: Use the create() method from the session object. This method returns a new FlowFile object, which you can
perform further processing on

Examples:

Groovy

groovy

flowFile = session.create()
// Additional processing here

Create a new FlowFile from a parent FlowFile

Use Case: You want to generate new FlowFile(s) based on an incoming FlowFile.

Approach: Use the create(parentFlowFile) method from the session object. This method takes a parent FlowFile
reference and returns a new child FlowFile object. The newly created FlowFile will inherit all the parent's attributes
except for the UUID. This method will automatically generate a Provenance FORK event or a Provenance JOIN event,
depending on whether other FlowFiles are generated from the same parent before the ProcessSession is committed.

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
newFlowFile = session.create(flowFile)
// Additional processing here

Add an attribute to a FlowFile

Use Case: You have a FlowFile to which you'd like to add a custom attribute.

Approach: Use the putAttribute(flowFile, attributeKey, attributeValue) method from the session object. This
method updates the given FlowFile's attributes with the given key/value pair. NOTE: The "uuid" attribute is fixed for a
FlowFile and cannot be modified; if the key is named "uuid", it will be ignored.

Also this is a good point to mention that FlowFile objects are immutable; this means that if you update a FlowFile's
attributes (or otherwise alter it) via the API, you will get a new reference to the new version of the FlowFile. This is
very important when it comes to transferring FlowFiles to relationships. You must keep a reference to the latest version
of a FlowFile, and you must transfer or remove the latest version of all FlowFiles retrieved from or created by the
session, otherwise you will get an error when executing. Most often, the variable used to store a FlowFile reference
will be overwritten with the latest version returned from a method that alters the FlowFile (intermediate FlowFile
references will be automatically discarded). In these examples you will see this technique of reusing a FlowFile
reference when adding attributes. Note that the current reference to the FlowFile is passed into the putAttribute()
method. The resulting FlowFile has an attribute named 'myAttr' with a value of 'myValue'. Also note that the method
takes a String for the value; if you have an Object you will have to serialize it to a String. Finally, please note that
if you are adding multiple attributes, it is better to create a Map and use putAllAttributes() instead (see next recipe
for details).

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')

Add multiple attributes to a FlowFile

Use Case: You have a FlowFile to which you'd like to add custom attributes.

Approach: Use the putAllAttributes(flowFile, attributeMap) method from the session object. This method updates
the given FlowFile's attributes with the key/value pairs from the given Map. NOTE: The "uuid" attribute is fixed for a
FlowFile and cannot be modified; if the key is named "uuid", it will be ignored.

Examples:

Groovy

groovy

attrMap = ['myAttr1': '1', 'myAttr2': Integer.toString(2)]
flowFile = session.get()
if (!flowFile) return
flowFile = session.putAllAttributes(flowFile, attrMap)

Get an attribute from a FlowFile

Use Case: You have a FlowFile from which you'd like to inspect an attribute.

Approach: Use the getAttribute(attributeKey) method from the FlowFile object. This method returns the String value
for the given attributeKey, or null if the attributeKey is not found. The examples show the retrieval of the value for
the "filename" attribute.

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
myAttr = flowFile.getAttribute('filename')

Get all attributes from a FlowFile

Use Case: You have a FlowFile from which you'd like to retrieve its attributes.

Approach: Use the getAttributes() method from the FlowFile object. This method returns a Map with String keys and
String values, representing the key/value pairs of attributes for the FlowFile. The examples show an iteration over the
Map of all attributes for a FlowFile.

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
flowFile.getAttributes().each { key, value ->
// Do something with the key/value pair
}

Transfer a FlowFile to a relationship

Use Case: After processing a FlowFile (new or incoming), you want to transfer the FlowFile to a relationship ("
success" or "failure"). In this simple case let us assume there is a variable called "errorOccurred" that indicates
which relationship to which the FlowFile should be transferred. Additional error handling techniques will be discussed
in part 2 of this series.

Approach: Use the transfer(flowFile, relationship) method from the session object. From the documentation: this
method transfers the given FlowFile to the appropriate destination processor work queue(s) based on the given
relationship. If the relationship leads to more than one destination the state of the FlowFile is replicated such that
each destination receives an exact copy of the FlowFile though each will have its own unique identity.

NOTE: ExecuteScript will perform a session.commit() at the end of each execution to ensure the operations have been
committed. You do not need to (and should not) perform a session.commit() within the script.

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
// Processing occurs here
if (errorOccurred) {
    session.transfer(flowFile, REL \ _FAILURE)
} else {
    session.transfer(flowFile, REL \ _SUCCESS)
}

Send a message to the log at a specified logging level

Use Case: You want to report some event that has occurred during processing to the logging framework.

Approach: Use the log variable with the warn(), trace(), debug(), info(), or error() methods. These methods can take
a single String, or a String followed by an array of Objects, or a String followed by an array of Objects followed by a
Throwable. The first one is used for simple messages. The second is used when you have some dynamic objects/values that
you want to log. To refer to these in the message string use "{}" in the message. These are evaluated against the Object
array in order of appearance, so if the message reads "Found these things: {} {} {}" and the Object array is ['Hello'
,1,true], then the logged message will be "Found these things: Hello 1 true". The third form of these logging methods
also takes a Throwable parameter, and is useful when an exception is caught and you want to log it.

Examples:

Groovy

groovy

log.info('Found these things: {} {} {}', \ ['Hello', 1, true \] as Object \ [\])

Read the contents of an incoming FlowFile using a callback

Use Case: You have incoming connection(s) to ExecuteScript and want to retrieve the contents of a FlowFile from the
queue(s) for processing.

Approach: Use the read(flowFile, inputStreamCallback) method from the session object. An InputStreamCallback
object is needed to pass into the read() method. Note that because InputStreamCallback is an object, the contents are
only visible to that object by default. If you need to use the data outside the read() method, use a more
globally-scoped variable. The examples will store the full contents of the incoming FlowFile into a String (using Apache
Commons' IOUtils class). NOTE: For large FlowFiles, this is not the best technique; rather you should read in only as
much data as you need, and process that as appropriate. For something like SplitText, you could read in a line at a time
and process it within the InputStreamCallback, or use the session.read(flowFile) approach mentioned earlier to get an
InputStream reference to use outside the callback.

Examples:

Groovy

groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = ''
// Cast a closure with an inputStream parameter to InputStreamCallback
session.read(flowFile, { inputStream ->
    text = IOUtils.toString(inputStream, StandardCharsets.UTF \ _8)
// Do something with text here
} as InputStreamCallback)

Write content to an outgoing FlowFile using a callback

Use Case: You want to generate content for an outgoing FlowFile.

Approach: Use the write(flowFile, outputStreamCallback) method from the session object. An OutputStreamCallback
object is needed to pass into the write() method. Note that because OutputStreamCallback is an object, the contents are
only visible to that object by default. If you need to use the data outside the write() method, use a more
globally-scoped variable. The examples will write a sample String to a FlowFile.

Examples:

Groovy

groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = 'Hello world!'
// Cast a closure with an outputStream parameter to OutputStreamCallback
flowFile = session.write(flowFile, { outputStream ->
    outputStream.write(text.getBytes(StandardCharsets.UTF \ _8))
} as OutputStreamCallback)

Overwrite an incoming FlowFile with updated content using a callback

Use Case: You want to reuse the incoming FlowFile but want to modify its content for the outgoing FlowFile.

Approach: Use the write(flowFile, streamCallback) method from the session object. An StreamCallback object is
needed to pass into the write() method. StreamCallback provides both an InputStream (from the incoming FlowFile) and an
outputStream (for the next version of that FlowFile), so you can use the InputStream to get the current contents of the
FlowFile, then modify them and write them back out to the FlowFile. This overwrites the contents of the FlowFile, so for
append you'd have to handle that by appending to the read-in contents, or use a different approach (with
session.append() rather than session.write() ). Note that because StreamCallback is an object, the contents are only
visible to that object by default. If you need to use the data outside the write() method, use a more globally-scoped
variable. The examples will reverse the contents of the incoming flowFile (assumed to be a String) and write out the
reversed string to a new version of the FlowFile.

Examples:

Groovy

groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = 'Hello world!'
// Cast a closure with an inputStream and outputStream parameter to StreamCallback
flowFile = session.write(flowFile, { inputStream, outputStream ->
    text = IOUtils.toString(inputStream, StandardCharsets.UTF \ _8)
    outputStream.write(text.reverse().getBytes(StandardCharsets.UTF \ _8))
} as StreamCallback)
session.transfer(flowFile, REL \ _SUCCESS)

Handle errors during script processing

Use Case: An error occurs in the script (either by data validation or a thrown exception), and you want the script
to handle it gracefully.

Approach: For exceptions, use the exception-handling mechanism for the scripting language (often they are try/catch
block(s)). For data validation, you can use a similar approach, but define a boolean variable like "valid" and an
if/else clause rather than a try/catch clause. ExecuteScript defines "success" and "failure" relationships; often your
processing will transfer "good" FlowFiles to success and "bad" FlowFiles to failure (logging an error in the latter
case).

Examples:

Groovy

groovy

flowFile = session.get()
if (!flowFile) return
try {
// Something that might throw an exception here

// Last operation is transfer to success (failures handled in the catch block)
    session.transfer(flowFile, REL \ _SUCCESS)
} catch (e) {
    log.error('Something went wrong', e)
    session.transfer(flowFile, REL \ _FAILURE)
}

Nifi Extension Bundles Nifi Scripting Bundle Nifi Scripting Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Script.ScriptedFilterRecord AdditionalDetails (`nifi-extension-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ScriptedFilterRecord/additionalDetails.md`)

ScriptedFilterRecord

Description

The ScriptedFilterRecord Processor provides the ability to use a scripting language, such as Groovy in order to remove
Records from an incoming FlowFile. NiFi provides several different Processors that can be used to work with Records in
different ways. Each of these processors has its pros and cons. The ScriptedFilterRecord is intended to work together
with these processors and be used as a pre-processing step before processing the FlowFile with more performance
consuming Processors, like ScriptedTransformRecord.

The Processor expects a user defined script in order to determine which Records should be kept and filtered out. When
creating a script, it is important to note that, unlike ExecuteScript, this Processor does not allow the script itself
to expose Properties to be configured or define Relationships.

This Processor maintains a Counter: "Records Processed" indicating the number of Records that were passed to the script
regardless of the result of the filtering.

Variable Bindings

While the script provided to this Processor does not need to provide boilerplate code or implement any
classes/interfaces, it does need some way to access the Records and other information that it needs in order to perform
its task. This is accomplished by using Variable Bindings. Each time that the script is invoked, each of the following
variables will be made available to the script:

Variable Name	Description	Variable Class
record	The Record that is to be processed.	Record
recordIndex	The zero-based index of the Record in the FlowFile.	Long (64-bit signed integer)
log	The Processor's Logger. Anything that is logged to this logger will be written to the logs as if the Processor itself had logged it. Additionally, a bulletin will be created for any log message written to this logger (though by default, the Processor will hide any bulletins with a level below WARN).	ComponentLog
attributes	Map of key/value pairs that are the Attributes of the FlowFile. Both the keys and the values of this Map are of type String. This Map is immutable. Any attempt to modify it will result in an UnsupportedOperationException being thrown.	java.util.Map

Return Value

Each time the script is invoked, it is expected to return a boolean value. Return values other than boolean,
including null value will be handled as unexpected script behaviour and handled accordingly: the processing will be
interrupted and the incoming FlowFile will be transferred to the failure relationship without further execution.

Example Scripts

Filtering based on position

The following script will keep only the first 2 Records from a FlowFile and filter out all the rest.

Example Input (CSV):

text

name, allyOf Decelea, Athens Corinth, Sparta Mycenae, Sparta Potidaea, Athens

Example Output (CSV):

text

name, allyOf Decelea, Athens Corinth, Sparta

Example Script (Groovy):

groovy

return recordIndex < 2 ? true : false

Filtering based on Record contents

The following script will filter the Records based on their content. Any Records satisfies the condition will be part of
the FlowFile routed to the success Relationship.

Example Input (JSON):

json

[
  {
    "city": "Decelea",
    "allyOf": "Athens"
  },
  {
    "city": "Corinth",
    "allyOf": "Sparta"
  },
  {
    "city": "Mycenae",
    "allyOf": "Sparta"
  },
  {
    "city": "Potidaea",
    "allyOf": "Athens"
  }
]

Example Output (CSV):

json

[
  {
    "city": "Decelea",
    "allyOf": "Athens"
  },
  {
    "city": "Potidaea",
    "allyOf": "Athens"
  }
]

Example Script (Groovy):

groovy

if (record.getValue("allyOf") == "Athens") {
    return true;
} else {
    return false;
}

Nifi Extension Bundles Nifi Scripting Bundle Nifi Scripting Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Script.ScriptedPartitionRecord AdditionalDetails (`nifi-extension-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ScriptedPartitionRecord/additionalDetails.md`)

ScriptedPartitionRecord

Description

The ScriptedPartitionRecord provides the ability to use a scripting language, such as Groovy, to quickly and easily
partition a Record based on its contents. There are multiple ways to reach the same behaviour such as using
PartitionRecord but working with user provided scripts opens a wide range of possibilities on the decision logic of
partitioning the individual records.

The provided script is evaluated once for each Record that is encountered in the incoming FlowFile. Each time that the
script is invoked, it is expected to return an object or a null value. The string representation of the return value
is used as the record's "partition". The null value is handled separately without conversion into string. All Records
with the same partition then will be batched to one FlowFile and routed to the success Relationship.

This Processor maintains a Counter with the name of "Records Processed". This represents the number of processed Records
regardless of partitioning.

Variable Bindings

Variable Name	Description	Variable Class
record	The Record that is to be processed.	Record
recordIndex	The zero-based index of the Record in the FlowFile.	Long (64-bit signed integer)
log	The Processor's Logger. Anything that is logged to this logger will be written to the logs as if the Processor itself had logged it. Additionally, a bulletin will be created for any log message written to this logger (though by default, the Processor will hide any bulletins with a level below WARN).	ComponentLog
attributes	Map of key/value pairs that are the Attributes of the FlowFile. Both the keys and the values of this Map are of type String. This Map is immutable. Any attempt to modify it will result in an UnsupportedOperationException being thrown.	java.util.Map

Return Value

The script is invoked separately for each Record. It is acceptable to return any Object might be represented as string.
This string value will be used as the partition of the given Record. Additionally, the script may return null.

Example

The following script will partition the input on the value of the "stellarType" field.

Example Input (CSV):

text

starSystem, stellarType Wolf 359, M Epsilon Eridani, K Tau Ceti, G Groombridge 1618, K Gliese 1, M

Example Output 1 (CSV) - for partition "M":

text

starSystem, stellarType Wolf 359,M Gliese 1,M

Example Output 2 (CSV) - for partition "K":

text

starSystem, stellarType Epsilon Eridani,K Groombridge 1618,K

Example Output 3 (CSV) - for partition "G":

text

starSystem, stellarType Tau Ceti,G

Note: the order of the outgoing FlowFiles is not guaranteed.

Example Script (Groovy):

groovy

return record.getValue("stellarType")

Nifi Extension Bundles Nifi Scripting Bundle Nifi Scripting Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Script.ScriptedTransformRecord AdditionalDetails (`nifi-extension-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ScriptedTransformRecord/additionalDetails.md`)

ScriptedTransformRecord

Description

The ScriptedTransformRecord provides the ability to use a scripting language, such as Groovy, to quickly and easily
update the contents of a Record. NiFi provides several different Processors that can be used to manipulate Records in
different ways. Each of these processors has its pros and cons. The ScriptedTransformRecord is perhaps the most powerful
and most versatile option. However, it is also the most error-prone, as it depends on writing custom scripts. It is also
likely to yield the lowest performance, as processors and libraries written directly in Java are likely to perform
better than interpreted scripts.

When creating a script, it is important to note that, unlike ExecuteScript, this Processor does not allow the script
itself to expose Properties to be configured or define Relationships. This is a deliberate decision. If it is necessary
to expose such configuration, the ExecuteScript processor should be used instead. By not exposing these elements, the
script avoids the need to define a Class or implement methods with a specific method signature. Instead, the script can
avoid any boilerplate code and focus purely on the task at hand.

The provided script is evaluated once for each Record that is encountered in the incoming FlowFile. Each time that the
script is invoked, it is expected to return a Record object (See note below regarding Return Values).
That Record is then written using the configured Record Writer. If the script returns a null value, the Record will
not be written. If the script returns an object that is not a Record, the incoming FlowFile will be routed to the
failure relationship.

This processor maintains two Counters: "Records Transformed" indicating the number of Records that were passed to the
script and for which the script returned a Record, and "Records Dropped" indicating the number of Records that were
passed to the script and for which the script returned a value of null.

Variable Bindings

Variable Name	Description	Variable Class
record	The Record that is to be transformed.	Record
recordIndex	The zero-based index of the Record in the FlowFile.	Long (64-bit signed integer)
log	The Processor's Logger. Anything that is logged to this logger will be written to the logs as if the Processor itself had logged it. Additionally, a bulletin will be created for any log message written to this logger (though by default, the Processor will hide any bulletins with a level below WARN).	ComponentLog
attributes	Map of key/value pairs that are the Attributes of the FlowFile. Both the keys and the values of this Map are of type String. This Map is immutable. Any attempt to modify it will result in an UnsupportedOperationException being thrown.	java.util.Map

Return Value

Each time that the script is invoked, it is expected to return
a Record
object or a Collection of Record objects. Those Records are then written using the configured Record Writer. If the
script returns a null value, the Record will not be written. If the script returns an object that is not a Record or
Collection of Records, the incoming FlowFile will be routed to the failure relationship.

The Record that is provided to the script is mutable. Therefore, it is a common pattern to update the record object in
the script and simply return that same record object.

Note: Depending on the scripting language, a script with no explicit return value may return null or may return
the last value that was referenced. Because returning null will result in dropping the Record and a non-Record return
value will result in an Exception (and simply for the sake of clarity), it is important to ensure that the configured
script has an explicit return value.

Adding a New Fields

A very common usage of Record-oriented processors is to allow the Record Reader to infer its schema and have the Record
Writer inherit the Record's schema. In this scenario, it is important to note that the Record Writer will inherit the
schema of the first Record that it encounters. Therefore, if the configured script will add a new field to a Record, it
is important to ensure that the field is added to all Records (with a null value where appropriate).

See the Adding New Fields example for more details.

Performance Considerations

NiFi offers many different processors for updating records in various ways. While each of these has its own pros and
cons, performance is often an important consideration. It is generally the case that standard processors, such as
UpdateRecord, will perform better than script-oriented processors. However, this may not always be the case. For
situations when performance is critical, the best case is to test both approaches to see which performs best.

A simple 5-minute benchmark was done to analyze the difference in performance. The script used simply modifies one field
and return the Record otherwise unmodified. The results are shown below. Note that no specifics are given regarding
hardware, specifically because the results should not be used to garner expectations of absolute performance but rather
to show relative performance between the different options.

Processor	Script Used	Records processed in 5 minutes
UpdateAttribute	No Script. User-defined Property added with name /num and value 42	50.1 million
ScriptedTransformRecord - Using Language: Groovy	record.setValue("num", 42)record	18.9 million

Example Scripts

Remove First Record

The following script will remove the first Record from each FlowFile that it encounters.

Example Input (CSV):

text

name, num Mark, 42 Felicia, 3720 Monica, -3

Example Output (CSV):

text

name, num Felicia, 3720 Monica, -3

Example Script (Groovy):

groovy

return recordIndex == 0 ? null : record

Replace Field Value

The following script will replace any field in a Record if the value of that field is equal to the value of the "Value
To Replace" attribute. The value of that field will be replaced with whatever value is in the "Replacement Value"
attribute.

Example Input Content (JSON):

json

[
  {
    "book": {
      "author": "John Doe",
      "date": "01/01/1980"
    }
  },
  {
    "book": {
      "author": "Jane Doe",
      "date": "01/01/1990"
    }
  }
]

Example Input Attributes:

Attribute Name	Attribute Value
Value To Replace	Jane Doe
Replacement Value	Author Unknown

Example Output (JSON):

json

[
  {
    "book": {
      "author": "John Doe",
      "date": "01/01/1980"
    }
  },
  {
    "book": {
      "author": "Author Unknown",
      "date": "01/01/1990"
    }
  }
]

Example Script (Groovy):

groovy

def replace(rec) {
    rec.toMap().each { k, v ->
        // If the field value is equal to the attribute 'Value to Replace', then set the         
        // field value to the 'Replacement Value' attribute.         
        if (v?.toString()?.equals(attributes['Value to Replace'])) {
            rec.setValue(k, attributes['Replacement Value'])
        }

        // Call Recursively if the value is a Record         
        if (v instanceof org.apache.nifi.serialization.record.Record) {
            replace(v)
        }
    }
}

replace(record)
return record

Pass-through

The following script allows each Record to pass through without altering the Record in any way.

Example Input:

Example output:

Example Script (Groovy):

groovy

record

Adding New Fields

The following script adds a new field named "favoriteColor" to all Records. Additionally, it adds an "isOdd" field to
all even-numbered Records.

It is important that all Records have the same schema. Since we want to add an "isOdd" field to Records 1 and 3, the
schema for Records 0 and 2 must also account for this. As a result, we will add the field to all Records but use a null
value for Records that are not even. See Adding New Fields for more information.

Example Input Content (CSV):

text

name, favoriteFood John Doe, Spaghetti Jane Doe, Pizza Jake Doe, Sushi June Doe, Hamburger

Example Output (CSV):

text

name, favoriteFood, favoriteColor, isOdd John Doe, Spaghetti, Blue, Jane Doe, Pizza, Blue, true Jake Doe, Sushi, Blue, June Doe, Hamburger, Blue, true

Example Script (Groovy):

groovy

import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;

// Always set favoriteColor to Blue. // Because we are calling #setValue with a String as the field name, the field type will be inferred. record.setValue("favoriteColor", "Blue")  // Set the 'isOdd' field to true if the record index is odd. Otherwise, set the 'isOdd' field to `null`. // Because the value may be `null` for the first Record (in fact, it always will be for this particular case), // we need to ensure that the Record Writer's schema be given the correct type for the field. As a result, we will not call // #setValue with a String as the field name but rather will pass a RecordField as the first argument, as the RecordField // allows us to specify the type of the field. // Also note that `RecordField` and `RecordFieldType` are `import`ed above. record.setValue(new RecordField("isOdd", RecordFieldType.BOOLEAN.getDataType()), recordIndex % 2 == 1 ? true : null)  return record

Fork Record

The following script return each Record that it encounters, plus another Record, which is derived from the first, but
where the 'num' field is one less than the 'num' field of the input.

Example Input (CSV):

text

name, num Mark, 42 Felicia, 3720 Monica, -3

Example Output (CSV):

text

name, num Mark, 42 Mark, 41 Felicia, 3720 Felicia, 3719 Monica, -3 Monica, -4

Example Script (Groovy):

groovy

import org.apache.nifi.serialization.record.*

def derivedValues = new HashMap(record.toMap())
derivedValues.put('num', derivedValues['num'] - 1)
derived = new MapRecord(record.schema, derivedValues)
return [record, derived]

Nifi Extension Bundles Nifi Scripting Bundle Nifi Scripting Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Script.ScriptedValidateRecord AdditionalDetails (`nifi-extension-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ScriptedValidateRecord/additionalDetails.md`)

ScriptedValidateRecord

Description

The ScriptedValidateRecord Processor provides the ability to use a scripting language, such as Groovy or Jyton in order
to validate Records in an incoming FlowFile. The provided script will be evaluated against each Record in an incoming
FlowFile. Each of those records will then be routed to either the "valid" or "invalid" FlowFile. As a result, each
incoming FlowFile may be broken up into two individual FlowFiles (if some records are valid and some are invalid,
according to the script), or the incoming FlowFile may have all of its Records kept together in a single FlowFile,
routed to either "valid" or "invalid" (if all records are valid or if all records are invalid, according to the script).

The Processor expects a user defined script in order to determine the validity of the Records. When creating a script,
it is important to note that, unlike ExecuteScript, this Processor does not allow the script itself to expose Properties
to be configured or define Relationships.

The provided script is evaluated once for each Record that is encountered in the incoming FlowFile. Each time that the
script is invoked, it is expected to return a boolean value, which is used to determine if the given Record is valid
or not: For Records the script returns with a true value, the given Record is considered valid and will be included to
the outgoing FlowFile which will be routed to the valid Relationship. For false values the given Record will be
added to the FlowFile routed to the invalid Relationship. Regardless of the number of incoming Records the outgoing
Records will be batched. For one incoming FlowFile there could be no more than one FlowFile routed to the valid and
the invalid Relationships. In case of there are no valid or invalid Record presents there will be no transferred
FlowFile for the respected Relationship. In addition to this the incoming FlowFile will be transferred to the original
Relationship without change. If the script returns an object that is not considered as boolean, the incoming FlowFile
will be routed to the failure Relationship instead and no FlowFile will be routed to the valid or invalid
Relationships.

This Processor maintains a Counter: "Records Processed" indicating the number of Records were processed by the
Processor.

Variable Bindings

Variable Name	Description	Variable Class
record	The Record that is to be processed.	Record
recordIndex	The zero-based index of the Record in the FlowFile.	Long (64-bit signed integer)
log	The Processor's Logger. Anything that is logged to this logger will be written to the logs as if the Processor itself had logged it. Additionally, a bulletin will be created for any log message written to this logger (though by default, the Processor will hide any bulletins with a level below WARN).	ComponentLog
attributes	Map of key/value pairs that are the Attributes of the FlowFile. Both the keys and the values of this Map are of type String. This Map is immutable. Any attempt to modify it will result in an UnsupportedOperationException being thrown.	java.util.Map

Return Value

Example Scripts

Validating based on position

The following script will consider only the first 2 Records as valid.

Example Input (CSV):

text

company, numberOfTrains Boston & Maine Railroad, 3 Chesapeake & Ohio Railroad, 2 Pennsylvania Railroad, 4 Reading Railroad, 2

Example Output (CSV) - valid Relationship:

text

company, numberOfTrains Boston & Maine Railroad, 3 Chesapeake & Ohio Railroad, 2

Example Output (CSV) - invalid Relationship:

text

company, numberOfTrains Pennsylvania Railroad, 4 Reading Railroad, 2

Example Script (Groovy):

groovy

return recordIndex < 2 ? true : false

Validating based on Record contents

The following script will filter the Records based on their content. Any Records satisfies the condition will be part of
the FlowFile routed to the valid Relationship, others wil lbe routed to the invalid Relationship.

Example Input (JSON):

json

[
  {
    "company": "Boston & Maine Railroad",
    "numberOfTrains": 3
  },
  {
    "company": "Chesapeake & Ohio Railroad",
    "numberOfTrains": -1
  },
  {
    "company": "Pennsylvania Railroad",
    "numberOfTrains": 2
  },
  {
    "company": "Reading Railroad",
    "numberOfTrains": 4
  }
]

Example Output (CSV) - valid Relationship:

json

[
  {
    "company": "Boston & Maine Railroad",
    "numberOfTrains": 3
  },
  {
    "company": "Pennsylvania Railroad",
    "numberOfTrains": 2
  },
  {
    "company": "Reading Railroad",
    "numberOfTrains": 4
  }
]

Example Output (CSV) - invalid Relationship:

json

[
  {
    "company": "Chesapeake & Ohio Railroad",
    "numberOfTrains": -1
  }
]

Example Script (Groovy):

groovy

if (record.getValue("numberOfTrains").toInteger() >= 0) {
    return true;
} else {
    return false;
}

Nifi Extension Bundles Nifi Shopify Bundle Nifi Shopify Processors Src Main Resources Docs Org.Apache.Nifi.Processors.Shopify.GetShopify AdditionalDetails (`nifi-extension-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/resources/docs/org.apache.nifi.processors.shopify.GetShopify/additionalDetails.md`)

GetShopify

Setting Up a Custom App

Follow the Shopify tutorial to enable and create private apps,
set API Scopes and generate API tokens.

Incremental Loading

Some resources can be processed incrementally by NiFi. This means that only resources created or modified after the last
run time of the processor are displayed. The processor state can be reset in the context menu. The following list shows
which date-time fields are incremented for which resources.

Customers
- Customers: updated_at_min
Discounts
- Price Rules: updated_at_min
Inventory
- Inventory Levels: updated_at_min
Online Store
- Script Tags: updated_at_min
Orders
- Abandoned Checkouts: updated_at_min
- Draft Orders: updated_at_min
- Orders: updated_at_min
Product
- Custom Collections: updated_at_min
- Products: updated_at_min
- Smart Collections: updated_at_min
Sales Channels
- Product Listings: updated_at_min
Store Properties
- Shipping Zones: updated_at_min

Nifi Extension Bundles Nifi Site To Site Reporting Bundle Nifi Site To Site Reporting Task Src Main Resources Docs Org.Apache.Nifi.Reporting.SiteToSiteBulletinReportingTask AdditionalDetails (`nifi-extension-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/resources/docs/org.apache.nifi.reporting.SiteToSiteBulletinReportingTask/additionalDetails.md`)

SiteToSiteBulletinReportingTask

The Site-to-Site Bulletin Reporting Task allows the user to publish Bulletin events using the Site To Site protocol.
Note: only up to 5 bulletins are stored per component and up to 10 bulletins at controller level for a duration of up to
5 minutes. If this reporting task is not scheduled frequently enough some bulletins may not be sent.

Record writer

The user can define a Record Writer and directly specify the output format and data with the assumption that the input
schema is the following:

text

/* Detailed source-code truncated for AI context efficiency. */

Nifi Extension Bundles Nifi Site To Site Reporting Bundle Nifi Site To Site Reporting Task Src Main Resources Docs Org.Apache.Nifi.Reporting.SiteToSiteMetricsReportingTask AdditionalDetails (`nifi-extension-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/resources/docs/org.apache.nifi.reporting.SiteToSiteMetricsReportingTask/additionalDetails.md`)

SiteToSiteMetricsReportingTask

The Site-to-Site Metrics Reporting Task allows the user to publish NiFi's metrics (as in the Ambari reporting task) to
the same NiFi instance or another NiFi instance. This provides a great deal of power because it allows the user to make
use of all the different Processors that are available in NiFi in order to process or distribute that data.

Ambari format

There are two available output formats. The first one is the Ambari format as defined in the Ambari Metrics Collector
API which is a JSON with dynamic keys. If using this format you might be interested in the below Jolt specification to
transform the data.

json

[
  {
    "operation": "shift",
    "spec": {
      "metrics": {
        "*": {
          "metrics": {
            "*": {
              "$": "metrics.[#4].metrics.time",
              "@": "metrics.[#4].metrics.value"
            }
          },
          "*": "metrics.[&1].&"
        }
      }
    }
  }
]

This would transform the below sample:

json

{
  "metrics": [
    {
      "metricname": "jvm.gc.time.G1OldGeneration",
      "appid": "nifi",
      "instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
      "hostname": "localhost",
      "timestamp": "1520456854361",
      "starttime": "1520456854361",
      "metrics": {
        "1520456854361": "0"
      }
    },
    {
      "metricname": "jvm.thread_states.terminated",
      "appid": "nifi",
      "instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
      "hostname": "localhost",
      "timestamp": "1520456854361",
      "starttime": "1520456854361",
      "metrics": {
        "1520456854361": "0"
      }
    }
  ]
}

into:

json

{
  "metrics": [
    {
      "metricname": "jvm.gc.time.G1OldGeneration",
      "appid": "nifi",
      "instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
      "hostname": "localhost",
      "timestamp": "1520456854361",
      "starttime": "1520456854361",
      "metrics": {
        "time": "1520456854361",
        "value": "0"
      }
    },
    {
      "metricname": "jvm.thread_states.terminated",
      "appid": "nifi",
      "instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
      "hostname": "localhost",
      "timestamp": "1520456854361",
      "starttime": "1520456854361",
      "metrics": {
        "time": "1520456854361",
        "value": "0"
      }
    }
  ]
}

Record format

The second format is leveraging the record framework of NiFi so that the user can define a Record Writer and directly
specify the output format and data with the assumption that the input schema is the following:

text

/* Detailed source-code truncated for AI context efficiency. */

Nifi Extension Bundles Nifi Site To Site Reporting Bundle Nifi Site To Site Reporting Task Src Main Resources Docs Org.Apache.Nifi.Reporting.SiteToSiteProvenanceReportingTask AdditionalDetails (`nifi-extension-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/resources/docs/org.apache.nifi.reporting.SiteToSiteProvenanceReportingTask/additionalDetails.md`)

SiteToSiteProvenanceReportingTask

The Site-to-Site Provenance Reporting Task allows the user to publish all the Provenance Events from a NiFi instance
back to the same NiFi instance or another NiFi instance. This provides a great deal of power because it allows the user
to make use of all the different Processors that are available in NiFi in order to process or distribute that data. When
possible, it is advisable to send the Provenance data to a different NiFi instance than the one that this Reporting Task
is running on, because when the data is received over Site-to-Site and processed, that in and of itself will generate
Provenance events. As a result, there is a cycle that is created. However, the data is sent in batches (1,000 by
default). This means that for each batch of Provenance events that are sent back to NiFi, the receiving NiFi will have
to generate only a single event per component.

By default, when published to a NiFi instance, the Provenance data is sent as a JSON array. However, the user can define
a Record Writer and directly specify the output format and data with the assumption that the input schema is defined as
follows:

text

/* Detailed source-code truncated for AI context efficiency. */

Nifi Extension Bundles Nifi Site To Site Reporting Bundle Nifi Site To Site Reporting Task Src Main Resources Docs Org.Apache.Nifi.Reporting.SiteToSiteStatusReportingTask AdditionalDetails (`nifi-extension-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/resources/docs/org.apache.nifi.reporting.SiteToSiteStatusReportingTask/additionalDetails.md`)

SiteToSiteStatusReportingTask

The Site-to-Site Status Reporting Task allows the user to publish Status events using the Site To Site protocol. The
component type and name filter regexes form a union: only components matching both regexes will be reported. However,
all process groups are recursively searched for matching components, regardless of whether the process group matches the
component filters.

Record writer

The user can define a Record Writer and directly specify the output format and data with the assumption that the input
schema is the following:

text

/* Detailed source-code truncated for AI context efficiency. */

--- METRICS ---

Files Extracted: 101
Estimated Token Budget: ~88419 tokens
Canonical Reference: https://codewiki.google/github.com/apache/nifi

Technical Documentation: apache/nifi

Overview (README.md) Copy Section

Apache NiFi

Status Copy Section

Resources Copy Section

Contacts Copy Section

Community Copy Section

Features Copy Chapter

Requirements Copy Chapter

Platform Requirements Copy Section

Optional Dependencies Copy Section

Projects Copy Chapter

Getting Started Copy Chapter

Developing Copy Chapter

Building Copy Section

Binaries Copy Section

Running Copy Chapter

Starting Copy Section

Accessing Copy Section

License Copy Chapter

Export Control Copy Chapter

QueryAirtableTable

Description Copy Section

Personal Access Token Copy Section

API rate limit Copy Section

Metadata API Copy Section

ConsumeAMQP

Summary Copy Chapter

AMQP Properties Copy Chapter

Configuration Details Copy Chapter

PublishAMQP

Summary Copy Chapter

Where did my message go? Copy Chapter

AMQP Properties Copy Chapter

AMQP Message Headers Source Copy Section

Configuration Details Copy Chapter

Nifi Extension Bundles Nifi Asn1 Bundle Nifi Asn1 Services Src Main Resources Docs Org.Apache.Nifi.Jasn1.JASN1Reader AdditionalDetails (`nifi-extension-bundles/nifi-asn1-bundle/nifi-asn1-services/src/main/resources/docs/org.apache.nifi.jasn1.JASN1Reader/additionalDetails.md`) Copy Section

JASN1Reader

Summary Copy Chapter

Troubleshooting Copy Section

Additional Preprocessing Copy Section

ConsumeKinesis

IAM permissions required for ConsumeKinesis Copy Chapter

Consumer Type Copy Chapter

Record processing Copy Chapter

Schema changes Copy Section

Output Strategies Copy Section

AWSSecretsManagerParameterProvider

Mapping AWS Secrets to Parameter Contexts Copy Section

Plain Text Secrets Copy Section

Configuring the Parameter Provider Copy Section

PutDynamoDBRecord

Description Copy Chapter

Data types Copy Chapter

Limitations Copy Chapter

Retry Copy Chapter

Partition and sort keys Copy Chapter

Partition Key Strategies Copy Section

Partition By Field

Partition By Attribute

Generated UUID

Sort Key Strategies Copy Section

None

Sort By Field

Generate Sequence

Examples Copy Chapter

Using fields as partition and sort key Copy Section

Setup

Result

Input

Output (stylized)

Using FlowFile filename as partition key with generated sort key Copy Section

Setup

Result

Input

Output (stylized)

First Item

Second Item

Using generated partition key Copy Section

Setup

Overview (README.md)

Status

Resources

Contacts

Community

Features

Requirements

Platform Requirements

Optional Dependencies

Projects

Getting Started

Developing

Building

Binaries

Running

Starting

Accessing

License

Export Control

Description

Personal Access Token

API rate limit

Metadata API

Summary

AMQP Properties

Configuration Details

Summary

Where did my message go?

AMQP Properties

AMQP Message Headers Source

Configuration Details

Nifi Extension Bundles Nifi Asn1 Bundle Nifi Asn1 Services Src Main Resources Docs Org.Apache.Nifi.Jasn1.JASN1Reader AdditionalDetails (`nifi-extension-bundles/nifi-asn1-bundle/nifi-asn1-services/src/main/resources/docs/org.apache.nifi.jasn1.JASN1Reader/additionalDetails.md`)

Summary

Troubleshooting

Additional Preprocessing

IAM permissions required for ConsumeKinesis

Consumer Type

Record processing

Schema changes

Output Strategies

Mapping AWS Secrets to Parameter Contexts

Plain Text Secrets

Configuring the Parameter Provider

Description

Data types

Limitations

Retry

Partition and sort keys

Partition Key Strategies

Sort Key Strategies

Examples

Using fields as partition and sort key

Using FlowFile filename as partition key with generated sort key

Using generated partition key

Streaming Versus Batch Processing

Per-Message Use Case

Per-Batch Use Case

GetAwsPollyJobStatus

Usage

StartAwsPollyJob

Usage

GetAwsTextractJobStatus

Usage

StartAwsTextractJob

Usage

DocumentAnalysis

ExpenseAnalysis

StartDocumentTextDetection

Usage

Usage

GetAwsTranslateJobStatus

Usage

StartAwsTranslateJob

Usage

Summary

Usage

Example Setup

PostgreSQL role and privileges

IAM permissions

CLI verification