Developing a NiFi Custom Processor with Java

Developing a custom processor in Apache NiFi is not merely a matter of extending AbstractProcessor and adding a few lines to onTrigger(). A processor participates in NiFi's scheduling, transaction, content repository, backpressure, clustering, and data provenance mechanisms. Compatibility with these mechanisms is as important as the correctness of the code itself.

A FlowFile consists of two separate parts. Content holds the actual data body, while attributes carry context about that data. Fields such as filename, path, and absolute.path are not data content. They are attributes used in routing and storage decisions. A processor does not modify a FlowFile directly. Every change creates a new FlowFile version through ProcessSession.

In the processors I developed for internal data flows, the decisive issue was placing the algorithm correctly within NiFi's execution model. When source-directory scanning, file selection, content validation, database transfer, and archiving were all placed in one method, the code could still run, but failure boundaries became unclear. A robust design required assigning one measurable responsibility to each processor.

Component contract

A Custom Processor is not exposed in the user interface merely as a Java class. Its name, description, properties, relationships, input requirement, and runtime behavior together form a component contract.

PropertyDescriptor objects define the processor's configuration fields. The source directory, timeout, batch size, validation option, and Controller Service connections are declared here. Every property should be constrained with an appropriate validator. Settings that are valid individually but incompatible in combination require customValidate(). NiFi does not allow a processor with invalid configuration to run.

Relationships carry an operation result to the next layer of the flow. success and failure may appear sufficient for many processors, but separate relationships are more appropriate when different error classes require different behavior. A retryable network error and structurally invalid data, for example, should not be routed to the same queue. The first can be sent to a delayed retry path, while the second can be routed to quarantine or inspection.

Using @InputRequirement(INPUT_FORBIDDEN) on source processors prevents an input connection from being created accidentally. For transformers that operate on an existing FlowFile, INPUT_REQUIRED is a more accurate contract. NiFi uses these annotations not only as documentation but also to determine component validity and which connections can be created in the interface.

The processor and its dependencies are distributed in a NAR package. A NAR isolates components and dependencies under a separate class loader. Copying an ordinary JAR into the NiFi lib directory is therefore not a durable plugin architecture. Shared Controller Service APIs should reside in a separate parent NAR, and the implementation and processor NARs should depend on the same API definition. Otherwise, types can become incompatible because their class-loader identities differ even when their class names appear identical.

Life cycle

The NiFi processor life cycle should be used to separate expensive preparation from the hot data path.

init() or initialize() runs when the component is created. Information that does not change during the processor's lifetime, such as the processor identity, can be obtained here. Runtime configuration may not yet be final, so configuration values should not be read at this stage.

@OnScheduled is called whenever the processor is started or rescheduled. Resolving the source directory to an absolute path, compiling regular expressions, building constant tables, preparing an immutable file list, and initializing a connection-pool object can be performed here. The framework guarantees that no other thread is concurrently executing processor code during this phase. If the prepared state is converted to an immutable object and published through a volatile reference, onTrigger() calls can read it without locking.

The source directory should not be taken from the FlowFile's absolute.path attribute. A source processor has not produced a FlowFile yet. The directory path should be read from a property, normalized, and converted to an absolute path when necessary. filename and relative.path later carry the generated FlowFile's location within the source. Confusing configuration with data attributes produces ambiguous behavior after restarts and under different working directories.

onTrigger() is the main data-processing path. Performing long preparation, class scanning, or rebuilding structures that do not change on every call increases latency. A source processor creates a new FlowFile. A transformation processor obtains an input with session.get() and returns immediately when no data is available.

@OnUnscheduled represents removal from scheduling, while @OnStopped represents cleanup after active invocations complete. Sockets, clients, executors, or local pools should be closed here. @OnShutdown provides an additional safeguard, but it is not guaranteed to run when the operating system or JVM terminates forcibly. Persistent correctness cannot depend only on a shutdown method.

FlowFile and transaction model

Because a FlowFile is immutable, the reference returned by every mutation call must be used:

flowFile = session.putAttribute(flowFile, "result", "success"); flowFile = session.write(flowFile, callback); session.transfer(flowFile, SUCCESS);

Continuing with an old reference can cause use of an invalid version within the session or unexpected routing errors.

ProcessSession groups FlowFile creation, reading, writing, cloning, removal, and transfer into one atomic unit of work. When the session commits, changes enter the persistent flow state. A rollback can be applied after a failure. ProcessSession is not thread-safe and must not be passed to worker threads outside onTrigger(). If external parallelism is required, session-independent data should first be converted into independent structures, and results should be applied to the session again on the thread executing the invocation.

The fundamental choice for content processing is a stream-based design. session.read(), session.write(), and StreamCallback make it possible to process large data without loading all of it into the heap. Because NiFi processors can run with several concurrent tasks, loading each FlowFile in full produces approximately the following memory demand:

instantaneous memory = concurrent task count x average content size

With eight concurrent tasks, reading 500 MB files into byte arrays requires about 4 GB of heap for the data bodies alone. Copies and parser objects are not included. The official developer guide likewise recommends avoiding full in-memory loading when content size is not strictly bounded.

The transaction boundary with an external system is more difficult. If a source file will be deleted after NiFi ingests it, the session should first commit safely. If the file is deleted first and NiFi stops before commit, data is lost. If NiFi stops after commit but before deletion from the external source, the same data can be ingested again. Exactly-once processing cannot be guaranteed across this interval. In most data-collection systems, controlled duplication is preferable to data loss.

Duplicate detection should therefore use a source identity, size, last-modified time, and, where necessary, a content digest. MD5 is fast, but it should not be used for validation where security or deliberate collisions matter. SHA-256 provides a stronger content identity, although reading a large file a second time creates I/O cost. If the filename and metadata are sufficient, calculating a hash for every flow is unnecessary.

Concurrency and memory management

NiFi can invoke onTrigger() on the same processor instance from multiple threads. @TriggerSerially reduces the concurrent task count to one, but it does not guarantee that every call executes on the same thread. All mutable state stored in processor fields must still be thread-safe.

A plain int counter and mutable list should not be used for ordered selection from a shared source directory. A fixed source array can be immutable or published through a volatile reference. AtomicInteger is sufficient for a round-robin index. A bounded BlockingQueue is more appropriate when items are borrowed and then returned to a pool.

One shared byte[] creates a data race across concurrent onTrigger() calls. For fixed, medium-sized buffers, ThreadLocal<byte[]> can reduce repeated allocation cost. For large buffers, a bounded buffer pool should be used to prevent memory consumption from growing disproportionately with the thread count. When the pool is empty, the call should wait, continue in smaller chunks, or fail in a controlled way instead of allocating unlimited new buffers.

ThreadLocal is not without limits. NiFi scheduler threads are long-lived. If very large buffers remain in a ThreadLocal, they stay attached to the heap even while the processor is idle. Buffer size and concurrent task count must be calculated together.

A processor should not create its own executor. An external thread pool can bypass NiFi's concurrent-task, backpressure, and shutdown model. If a required protocol client uses worker threads internally, its life cycle must be managed explicitly between @OnScheduled and @OnStopped.

Cluster and failure behavior

A source processor that is correct on one node can read the same file on every node in a cluster. For processors that scan a remote or shared file system, @PrimaryNodeOnly can prevent these duplicates. If greater parallelism is required, file ownership, atomic claiming, or cluster-wide coordination must be designed.

StateManager is suitable for a last-processed identity, timestamp, or small checkpoint. Scope.LOCAL keeps separate state on each node. Scope.CLUSTER exposes shared state to all nodes. State Manager is intended for small key-value data and should not be used as a high-volume record store. The cluster-state map has a size limit. Large or frequently updated state should move to an external data store or a shared Controller Service layer.

Backpressure is not merely a flow-designer setting. It works together with the processor's failure behavior. When an output queue is full, NiFi normally stops scheduling the processor and propagates pressure upstream. If the processor bypasses this with its own thread or an unbounded internal queue, NiFi's protection becomes ineffective.

Penalization delays the time at which a particular FlowFile becomes available again. Yield prevents the whole processor from being scheduled for a short period. A FlowFile can be penalized for a temporary data-specific error. If an entire remote system is unavailable, context.yield() may be more appropriate. When an expected error escapes as ProcessException, the framework rolls the session back and penalizes the relevant FlowFiles. An administrative yield is also applied after unexpected runtime failures.

Retries should not be unlimited. Attempt count can be tracked in an attribute or persistent state. A short deterministic timeout, increasing delay, and a maximum limit should be used together. For a persistently unavailable external source, a circuit breaker prevents the same expensive connection attempt on every trigger.

Testing and production deployment

The NiFi Mock Framework executes processor and Controller Service tests in a manner close to the actual life cycle. TestRunner can validate property configuration, the FlowFile queue, relationships, content, attribute values, and StateManager behavior. Calls to @OnScheduled, onTrigger, @OnUnscheduled, and @OnStopped can be included in the test flow.

Unit tests should cover at least the following cases:

• Empty input
• Partial and corrupted content
• Reappearance of the same file
• A full output queue
• Remote-system timeout
• Session rollback
• Multiple concurrent tasks
• Difference between local and cluster state
• Closure of open resources while the processor stops
• NFS or file-system write failure

A long-running load test is required before production deployment. Average duration alone is insufficient. P95 and P99 processing time, FlowFile queue length, repository I/O, heap use, garbage-collection pauses, open file descriptors, and retry rate should be observed together.

The processor should log activity through ComponentLog and report retrieval from an external source or transmission to an external target through ProvenanceReporter. Provenance is not merely a debugging record. It is part of the chain that shows where data originated, which transformations it underwent, and where it was sent.

A Custom Processor can be deployed in a critical system. The requirement is not merely that it work on the normal path. It should produce deterministic results for the same input, carry no state exposed to thread races, process large data as a stream, and avoid data loss during failure. When source collection, content processing, database transfer, and archiving are divided among separate processors, retry, backpressure, and observability boundaries can be managed independently for each stage.

NiFi's value is not that it eliminates custom code. It runs custom code within transaction, queue, provenance, and life-cycle rules. A well-written Custom Processor does not attempt to bypass these rules. It aligns its algorithm with NiFi's data-flow model.