NiFi Developer’s Guide

In order to understand the Processor API, we must first understand - at least at a high level - several supporting classes and interfaces, which are discussed below.

Since the vast majority of Processors will be created by extending the AbstractProcessor, it is the abstract class that we will examine in this section. The AbstractProcessor provides several methods that will be of interest to Processor developers.

The NiFi API provides lifecycle support through use of Java Annotations. The org.apache.nifi.annotations.lifecycle package contains several annotations for lifecycle management. The following Annotations may be applied to Java methods in a NiFi component to indicate to the framework when the methods should be called. For the discussion of Component Lifecycle, we will define a NiFi component as a Processor, ControllerServices, or ReportingTask.

Processors are responsible for reporting their activity so that users are able to understand what happens to their data. Processors should log events via the ProcessorLog, which is accessible via the InitializationContext or by calling the getLogger method of AbstractProcessor.

Additionally, Processors should use the ProvenanceReporter interface, obtained via the ProcessSession’s getProvenanceReporter method. The ProvenanceReoprter should be used to indicate any time that content is received from an external source or sent to an external location. The ProvenanceReporter also has methods for reporting when a FlowFile is cloned, forked, or modified, and when multiple FlowFiles are merged into a single FlowFile as well as associating a FlowFile with some other identifier. However, these functions are less critical to report, as the framework is able to detect these things and emit appropriate events on the Processor’s behalf. Yet, it is a best practice for the Processor developer to emit these events, as it becomes explicit in the code that these events are being emitted, and the developer is able to provide additional details to the events, such as the amount of time that the action took or pertinent information about the action that was taken. If the Processor emits an event, the framework will not emit a duplicate event. Instead, it always assumes that the Processor developer knows what is happening in the context of the Processor better than the framework does. The framework may, however, emit a different event. For example, if a Processor modifies both the content of a FlowFile and its attributes and then emits only an ATTRIBUTES_MODIFIED event, the framework will emit a CONTENT_MODIFIED event. The framework will not emit an ATTRIBUTES_MODIFIED event if any other event is emitted for that FlowFile (either by the Processor or the framework). This is due to the fact that all Provenance Events know about the attributes of the FlowFile before the event occurred as well as those attributes that occurred as a result of the processing of that FlowFile, and as a result the ATTRIBUTES_MODIFIED is generally considered redundant and would result in a rendering of the FlowFile lineage being very verbose. It is, however, acceptable for a Processor to emit this event along with others, if the event is considered pertinent from the perspective of the Processor.

Individual properties can be documented by calling the description method of a PropertyDescriptor’s builder as such:

If the property is to provide a set of allowable values, those values are presented to the user in a drop-down field in the UI. Each of those values can also be given a description:

Processor Relationships are documented in much the same way that properties are - by calling the description method of a Relationship’s builder:

The org.apache.nifi.annotations.documentation package provides Java annotations that can be used to document components. The CapabilityDescription annotation can be added to a Processor, Reporting Task, or Controller Service and is intended to provide a brief description of the functionality provided by the component. The Tags annotation has a value variable that is defined to be an Array of Strings. As such, it is used by providing multiple values as a comma-separated list of `String`s with curly braces. These values are then incorporated into the UI by allowing users to filter the components based on a tag (i.e., a keyword). Additionally, the UI provides a tag cloud that allows users to select the tags that they want to filter by. The tags that are largest in the cloud are those tags that exist the most on the components in that instance of NiFi. An example of using these annotations is provided below:

When the documentation methods above are not sufficient, NiFi provides the ability to expose more advanced documentation to the user via the "Usage" documentation. When a user right-clicks on a Processor, NiFi provides a "Usage" menu item in the context menu. Additionally, the UI exposes a "Help" link in the top-right corner, from which the same Usage information can be found.

The advanced documentation of a Processor is provided as an HTML file. This file should exist within a directory whose name is the fully-qualified name of the component, and this directory’s parent should be named docs and exist in the root of the Processor’s jar. The mechanism provided for this will be changing as of the 0.1.0 release. At that time, this section will be updated to reflect the new procedures for providing this advanced documentation.

A Processor that ingests data into NiFi has a single Relationship names success. This Processor generates new FlowFiles via the ProcessSession create method and does not pull FlowFiles from incoming Connections. The Processor name starts with "Get" or "Listen," depending on whether it polls an external source or exposes some interface to which external sources can connect. The name ends with the protocol used for communications. Processors that follow this pattern include GetFile, GetSFTP, ListenHTTP, and GetHTTP.

This Processor may create or initialize a Connection Pool in a method that uses the @OnScheduled annotation. However, because communications problems may prevent connections from being established or cause connections to be terminated, connections themselves are not created at this point. Rather, the connections are created or leased from the pool in the onTrigger method.

The onTrigger method of this Processor begins by leasing a connection from the Connection Pool, if possible, or otherwise creates a connection to the external service. When no data is available from the external source, the yield method of the ProcessContext is called by the Processor and the method returns so that this Processor avoids continually running and depleting resources without benefit. Otherwise, this Processor then creates a FlowFile via the ProcessSession’s create method and assigns an appropriate filename and path to the FlowFile (by adding the filename and path attributes), as well as any other attributes that may be appropriate. An OutputStream to the FlowFile’s content is obtained via the ProcessSession’s write method, passing a new OutputStreamCallback (which is usually an anonymous inner class). From within this callback, the Processor is able to write to the FlowFile and streams the content from the external resource to the FlowFile’s OutputStream. If the desire is to write the entire contents of an InputStream to the FlowFile, the importFrom method of ProcessSession may be more convenient to use than the write method.

When this Processor expects to receive many small files, it may be advisable to create several FlowFiles from a single session before committing the session. Typically, this allows the Framework to treat the content of the newly created FlowFiles much more efficiently.

This Processor generates a Provenance event indicating that it has received data and specifies from where the data came. This Processor should log the creation of the FlowFile so that the FlowFile’s origin can be determined by analyzing logs, if necessary.

This Processor acknowledges receipt of the data and/or removes the data from the external source in order to prevent receipt of duplicate files. This is done only after the ProcessSession by which the FlowFile was created has been committed! Failure to adhere to this principle may result in data loss, as restarting NiFi before the session has been committed will result in the temporary file being deleted. Note, however, that it is possible using this approach to receive duplicate data because the application could be restarted after committing the session and before acknowledging or removing the data from the external source. In general, though, potential data duplication is preferred over potential data loss. The connection is finally returned or added to the Connection Pool, depending on whether the connection was leased from the Connection Pool to begin with or was created in the onTrigger method.

If there is a communications problem, the connection is typically terminated and not returned (or added) to the Connection Pool. Connections to remote systems are torn down and the Connection Pool shutdown in a method annotated with the @OnStopped annotation so that resources can be reclaimed.

A Processor that publishes data to an external source has two Relationships: success and failure. The Processor name starts with "Put" followed by the protocol that is used for data transmission. Processors that follow this pattern include PutEmail, PutSFTP, and PostHTTP (note that the name does not begin with "Put" because this would lead to confusion, since PUT and POST have special meanings when dealing with HTTP).

This Processor may create or initialize a Connection Pool in a method that uses the @OnScheduled annotation. However, because communications problems may prevent connections from being established or cause connections to be terminated, connections themselves are not created at this point. Rather, the connections are created or leased from the pool in the onTrigger method.

The onTrigger method first obtains a FlowFile from the ProcessSession via the get method. If no FlowFile is available, the method returns without obtaining a connection to the remote resource.

If at least one FlowFile is available, the Processor obtains a connection from the Connection Pool, if possible, or otherwise creates a new connection. If the Processor is neither able to lease a connection from the Connection Pool nor create a new connection, the FlowFile is routed to failure, the event is logged, and the method returns.

If a connection was obtained, the Processor obtains an InputStream to the FlowFile’s content by invoking the read method on the ProcessSession and passing an InputStreamCallback (which is often an anonymous inner class) and from within that callback transmits the contents of the FlowFile to the destination. The event is logged along with the amount of time taken to transfer the file and the data rate at which the file was transferred. A SEND event is reported to the ProvenanceReporter by obtaining the reporter from the ProcessSession via the getProvenanceReporter method and calling the send method on the reporter. The connection is returned or added to the Connection Pool, depending on whether the connection was leased from the pool or newly created by the onTrigger method.

If there is a communications problem, the connection is typically terminated and not returned (or added) to the Connection Pool. If there is an issue sending the data to the remote resource, the desired approach for handling the error depends on a few considerations. If the issue is related to a network condition, the FlowFile is generally routed to failure. The FlowFile is not penalized because there is not necessary a problem with the data. Unlike the case of the Data Ingress Processor, we typically do not call yield on the ProcessContext. This is because in the case of ingest, the FlowFile does not exist until the Processor is able to perform its function. However, in the case of a Put Processor, the DataFlow Manager may choose to route failure to a different Processor. This can allow for a "backup" system to be used in the case of problems with one system or can be used for load distribution across many systems.

If a problem occurs that is data-related, one of two approaches should be taken. First, if the problem is likely to sort itself out, the FlowFile is penalized and then routed to failure. This is the case, for instance, with PutFTP, when a FlowFile cannot be transferred because of a file naming conflict. The presumption is that the file will eventually be removed from the directory so that the new file can be transferred. As a result, we penalize the FlowFile and route to failure so that we can try again later. In the other case, if there is an actual problem with the data (such as the data does not conform to some required specification), a different approach may be taken. In this case, it may be advantageous to break apart the failure relationship into a failure and a communications failure relationship. This allows the DataFlow Manager to determine how to handle each of these cases individually. It is important in these situations to document well the differences between the two Relationships by clarifying it in the "description" when creating the Relationship.

Connections to remote systems are torn down and the Connection Pool shutdown in a method annotated with @OnStopped so that resources can be reclaimed.

A Processor that routes data based on its content will take one of two forms: Route an incoming FlowFile to exactly one destination, or route incoming data to 0 or more destinations. Here, we will discuss the first case.

This Processor has two relationships: matched and unmatched. If a particular data format is expected, the Processor will also have a failure relationship that is used when the input is not of the expected format. The Processor exposes a Property that indicates the routing criteria.

If the Property that specifies routing criteria requires processing, such as compiling a Regular Expression, this processing is done in a method annotated with @OnScheduled, if possible. The result is then stored in a member variable that is marked as volatile.

The onTrigger method obtains a single FlowFile. The method reads the contents of the FlowFile via the ProcessSession’s read method, evaluating the Match Criteria as the data is streamed. The Processor then determines whether the FlowFile should be routed to matched or unmatched based on whether or not the criteria matched, and routes the FlowFile to the appropriate relationship.

The Processor then emits a Provenance ROUTE event indicating which Relationship to which the Processor routed the FlowFile.

This Processor is annotated with the @SideEffectFree and @SupportsBatching annotations from the org.apache.nifi.annotations.behavior package.

If a Processor will route a single FlowFile to potentially many relationships, this Processor will be slightly different than the above-described Processor for Routing Data Based on Content. This Processor typically has Relationships that are dynamically defined by the user as well as an unmatched relationship.

In order for the user to be able to define additionally Properties, the getSupportedDynamicPropertyDescriptor method must be overridden. This method returns a PropertyDescriptor with the supplied name and an applicable Validator to ensure that the user-specified Matching Criteria is valid.

In this Processor, the Set of Relationships that is returned by the getRelationships method is a member variable that is marked volatile. This Set is initially constructed with a single Relationship named unmatched. The onPropertyModified method is overridden so that when a Property is added or removed, a new Relationship is created with the same name. If the Processor has Properties that are not user-defined, it is important to check if the specified Property is user-defined. This can be achieved by calling the isDynamic method of the PropertyDescriptor that is passed to this method. If this Property is dynamic, a new Set of Relationships is then created, and the previous set of Relationships is copied into it. This new Set either has the newly created Relationship added to it or removed from it, depending on whether a new Property was added to the Processor or a Property was removed (Property removal is detected by check if the third argument to this function is null). The member variable holding the Set of Relationships is then updated to point to this new Set.

If the Properties that specify routing criteria require processing, such as compiling a Regular Expression, this processing is done in a method annotated with @OnScheduled, if possible. The result is then stored in a member variable that is marked as volatile. This member variable is generally of type Map where the key is of type Relationship and the value’s type is defined by the result of processing the property value.

The onTrigger method obtains a FlowFile via the get method of ProcessSession. If no FlowFile is available, it returns immediately. Otherwise, a Set of type Relationship is created. The method reads the contents of the FlowFile via the ProcessSession’s read method, evaluating each of the Match Criteria as the data is streamed. For any criteria that matches, the relationship associated with that Match Criteria is added to the Set of Relationships.

After reading the contents of the FlowFile, the method checks if the Set of Relationships is empty. If so, the original FlowFile has an attribute added to it to indicate the Relationship to which it was routed and is routed to the unmatched. This is logged, a Provenance ROUTE event is emitted, and the method returns. If the size of the Set is equal to 1, the original FlowFile has an attribute added to it to indicate the Relationship to which it was routed and is routed to the Relationship specified by the entry in the Set. This is logged, a Provenance ROUTE event is emitted for the FlowFile, and the method returns.

In the event that the Set contains more than 1 Relationship, the Processor creates a clone of the FlowFile for each Relationship, except for the first. This is done via the clone method of the ProcessSession. There is no need to report a CLONE Provenance Event, as the framework will handle this for you. The original FlowFile and each clone are routed to their appropriate Relationship with attribute indicating the name of the Relationship. A Provenance ROUTE event is emitted for each FlowFile. This is logged, and the method returns.

This Processor is annotated with the @SideEffectFree and @SupportsBatching annotations from the org.apache.nifi.annotations.behavior package.

The previous description of Route Based on Content (One-to-Many) provides an abstraction for creating a very powerful Processor. However, it assumes that each FlowFile will be routed in its entirety to zero or more Relationships. What if the incoming data format is a "stream" of many different pieces of information - and we want to send different pieces of this stream to different Relationships? For example, imagine that we want to have a RouteCSV Processor such that it is configured with multiple Regular Expressions. If a line in the CSV file matches a Regular Expression, that line should be included in the outbound FlowFile to the associated relationship. If a Regular Expression is associated with the Relationship "has-apples" and that Regular Expression matches 1,000 of the lines in the FlowFile, there should be one outbound FlowFile for the "has-apples" relationship that has 1,000 lines in it. If a different Regular Expression is associated with the Relationship "has-oranges" and that Regular Expression matches 50 lines in the FlowFile, there should be one outbound FlowFile for the "has-oranges" relationship that has 50 lines in it. I.e., one FlowFile comes in and two FlowFiles come out. The two FlowFiles may contain some of the same lines of text from the original FlowFile, or they may be entirely different. This is the type of Processor that we will discuss in this section.

This Processor’s name starts with "Route" and ends with the name of the data type that it routes. In our example here, we are routing CSV data, so the Processor is named RouteCSV. This Processor supports dynamic properties. Each user-defined property has a name that maps to the name of a Relationship. The value of the Property is in the format necessary for the "Match Criteria." In our example, the value of the property must be a valid Regular Expression.

This Processor maintains an internal ConcurrentMap where the key is a Relationship and the value is of a type dependent on the format of the Match Criteria. In our example, we would maintain a ConcurrentMap<Relationship, Pattern>. This Processor overrides the onPropertyModified method. If the new value supplied to this method (the third argument) is null, the Relationship whose name is defined by the property name (the first argument) is removed from the ConcurrentMap. Otherwise, the new value is processed (in our example, by calling Pattern.compile(newValue)) and this value is added to the ConcurrentMap with the key again being the Relationship whose name is specified by the property name.

This Processor will override the customValidate method. In this method, it will retrieve all Properties from the ValidationContext and count the number of PropertyDescriptors that are dynamic (by calling isDynamic() on the PropertyDescriptor). If the number of dynamic PropertyDescriptors is 0, this indicates that the user has not added any Relationships, so the Processor returns a ValidationResult indicating that the Processor is not valid because it has no Relationships added.

The Processor returns all of the Relationships specified by the user when its getRelationships method is called and will also return an unmatched Relationship. Because this Processor will have to read and write to the Content Repository (which can be relatively expensive), if this Processor is expected to be used for very high data volumes, it may be advantageous to add a Property that allows the user to specify whether or not they care about the data that does not match any of the Match Criteria.

When the onTrigger method is called, the Processor obtains a FlowFile via ProcessSession.get. If no data is available, the Processor returns. Otherwise, the Processor creates a Map<Relationship, FlowFile>. We will refer to this Map as flowFileMap. The Processor reads the incoming FlowFile by calling ProcessSession.read and provides an InputStreamCallback. From within the Callback, the Processor reads the first piece of data from the FlowFile. The Processor then evaluates each of the Match Criteria against this piece of data. If a particular criteria (in our example, a Regular Expression) matches, the Processor obtains the FlowFile from flowFileMap that belongs to the appropriate Relationship. If no FlowFile yet exists in the Map for this Relationship, the Processor creates a new FlowFile by calling session.create(incomingFlowFile) and then adds the new FlowFile to flowFileMap. The Processor then writes this piece of data to the FlowFile by calling session.append with an OutputStreamCallback. From within this OutputStreamCallback, we have access to the new FlowFile’s OutputStream, so we are able to write the data to the new FlowFile. We then return from the OutputStreamCallback. After iterating over each of the Match Criteria, if none of them match, we perform the same routines as above for the unmatched relationship (unless the user configures us to not write out unmatched data). Now that we have called session.append, we have a new version of the FlowFile. As a result, we need to update our flowFileMap to associate the Relationship with the new FlowFile.

If at any point, an Exception is thrown, we will need to route the incoming FlowFile to failure. We will also need to remove each of the newly created FlowFiles, as we won’t be transferring them anywhere. We can accomplish this by calling session.remove(flowFileMap.values()). At this point, we will log the error and return.

Otherwise, if all is successful, we can now iterate through the flowFileMap and transfer each FlowFile to the corresponding Relationship. The original FlowFile is then either removed or routed to an original relationship. For each of the newly created FlowFiles, we also emit a Provenance ROUTE event indicating which Relationship the FlowFile went to. It is also helpful to include in the details of the ROUTE event how many pieces of information were included in this FlowFile. This allows DataFlow Managers to easily see when looking at the Provenance Lineage view how many pieces of information went to each of the relationships for a given input FlowFile.

Additionally, some Processors may need to "group" the data that is sent to each Relationship so that each FlowFile that is sent to a relationship has the same value. In our example, we may wan to allow the Regular Expression to have a Capturing Group and if two different lines in the CSV match the Regular Expression but have different values for the Capturing Group, we want them to be added to two different FlowFiles. The matching value could then be added to each FlowFile as an Attribute. This can be accomplished by modifying the flowFileMap such that it is defined as Map<Relationship, Map<T, FlowFile>> where T is the type of the Grouping Function (in our example, the Group would be a String because it is the result of evaluating a Regular Expression’s Capturing Group).

This Processor is almost identical to the Route Data Based on Content Processors described above. It takes two different forms: One-to-One and One-to-Many, as do the Content-Based Routing Processors. This Processor, however, does not make any call to ProcessSession’s read method, as it does not read FlowFile content. This Processor is typically very fast, so the @SupportsBatching annotation can be very important in this case.

This Processor generally requires no user configuration, with the exception of the size of each Split to create. The onTrigger method obtains a FlowFile from its input queues. A List of type FlowFile is created. The original FlowFile is read via the ProcessSession’s read method, and an InputStreamCallback is used. Within the InputStreamCallback, the content is read until a point is reached at which the FlowFile should be split. If no split is needed, the Callback returns, and the original FlowFile is routed to success. In this case, a Provenance ROUTE event is emitted. Typically, ROUTE events are not emitted when routing a FlowFile to success because this generates a very verbose lineage that becomes difficult to navigate. However, in this case,the event is useful because we would otherwise expect a FORK event and the absence of any event is likely to cause confusion. The fact that the FlowFile was not split but was instead transferred to success is logged, and the method returns.

If a point is reached at which a FlowFile needs to be split, a new FlowFile is created via the ProcessSession’s create(FlowFile) method or the clone(FlowFile, long, long) method. The next section of code depends on whether the create method is used or the clone method is used. Both methods are described below. Which solution is appropriate must be determined on a case-by-case basis.

The Create Method is most appropriate when the data will not be directly copied from the original FlowFile to the new FlowFile. For example, if only some of the data will be copied, or if the data will be modified in some way before being copied to the new FlowFile, this method is necessary. However, if the content of the new FlowFile will be an exact copy of a portion of the original FlowFile, the Clone Method is much preferred.

Create Method If using the create method, the method is called with the original FlowFile as the argument so that the newly created FlowFile will inherit the attributes of the original FlowFile and a Provenance FORK event will be created by the framework.

The code then enters a try/finally block. Within the finally block, the newly created FlowFile is added to the List of FlowFiles that have been created. This is done within a finally block so that if an Exception is thrown, the newly created FlowFile will be appropriately cleaned up. Within the try block, the callback initiates a new callback by calling the ProcessSession’s write method with an OutputStreamCallback. The appropriate data is then copied from the InputStream of the original FlowFile to the OutputStream for the new FlowFile.

Clone Method If the content of the newly created created FlowFile is to be only a contiguous subset of the bytes of the original FlowFile, it is preferred to use the clone(FlowFile, long, long) method instead of the create(FlowFile) method of the ProcessSession. In this case, the offset of the original FlwoFile at which the new FlowFile’s content should begin is passed as the second argument to the clone method. The length of the new FlowFile is passed as the third argument to the clone method. For example, if the original FlowFile was 10,000 bytes and we called clone(flowFile, 500, 100), the FlowFile that would be returned to us would be identical to flowFile with respect to its attributes. However, the content of the newly created FlowFile would be 100 bytes in length and would start at offset 500 of the original FlowFile. That is, the contents of the newly created FlowFile would be the same as if you had copied bytes 500 through 599 of the original FlowFile.

After the clone has been created, it is added to the List of FlowFiles.

This method is much more highly preferred than the Create method, when applicable, because no disk I/O is required. The framework is able to simply create a new FlowFile that references a subset of the original FlowFile’s content, rather than actually copying the data. However, this is not always possible. For example, if header information must be copied from the beginning of the original FlowFile and added to the beginning of each Split, then this method is not possible.

Both Methods Regardless of whether the Clone Method or the Create Method is used, the following is applicable.

If at any point in the InputStreamCallback, a condition is reached in which processing cannot continue (for example, the input is malformed), a ProcessException should be thrown. The call to the ProcesssSession’s read method is wrapped in a try/catch block where ProcessException is caught. If an Exception is caught, a log message is generated explaining the error. The List of newly created FlowFiles is removed via the ProcessSession’s remove method. The original FlowFile is routed to failure.

If no problems arise, the original FlowFile is routed to original and all newly created FlowFiles are updated to include the following attributes:

The newly created FlowFiles are routed to success; this event is logged; and the method returns.

This Processor is very similar to the Route Based on Content Processors discussed above. Rather than routing a FlowFile to matched or unmatched, the FlowFile is generally routed to success or failure and attributes are added to the FlowFile as appropriate. The attributes to be added are configured in a manner similar to that of the Route Based on Content (One-to-Many), with the user defining their own properties. The name of the property indicates the name of an attribute to add. The value of the property indicates some Matching Criteria to be applied to the data. If the Matching Criteria matches the data, an attribute is added with the name the same as that of the Property. The value of the attribute is the criteria from the content that matched.

For example, a Processor that evaluates XPath Expressions may allow user-defined XPaths to be entered. If the XPath matches the content of a FlowFile, that FlowFile will have an attribute added with the name being equal to that of the Property name and a value equal to the textual content of the XML Element or Attribute that matched the XPath. The failure relationship would then be used if the incoming FlowFile was not valid XML in this example. The success relationship would be used regardless of whether or not any matches were found. This can then be used to route the FlowFile when appropriate.

This Processor emits a Provenance Event of type ATTRIBUTES_MODIFIED.

The Enrich/Modify Content pattern is very common and very generic. This pattern is responsible for any general content modification. For the majority of cases, this Processor is marked with the @SideEffectFree and @SupportsBatching annotations. The Processor has any number of required and optional Properties, depending on the Processor’s function. The Processor generally has a success and failure relationship. The failure relationship is generally used when the input file is not in the expected format.

This Processor obtains a FlowFile and updates it using the ProcessSession’s write(StreamCallback) method so that it is able to both read from the FlowFile’s content and write to the next version of the FlowFile’s content. If errors are encountered during the callback, the callback will throw a ProcessException. The call to the ProcessSession’s write method is wrapped in a try/catch block that catches ProcessException and routes the FlowFile to failure.

If the callback succeeds, a CONTENT_MODIFIED Provenance Event is emitted.

During the execution of the onTrigger method of a Processor, many things can potentially go awry. Common failure conditions include:

Any of these conditions can result in an Exception being thrown from the Processor. From the framework perspective, there are two types of Exceptions that can escape a Processor: ProcessException and all others.

If a ProcessException is thrown from the Processor, the framework will assume that this is a failure that is a known outcome. Moreover, it is a condition where attempting to process the data again later may be successful. As a result, the framework will roll back the session that was being processed and penalize the FlowFiles that were being processed.

If any other Exception escapes the Processor, though, the framework will assume that it is a failure that was not taken into account by the developer. In this case, the framework will also roll back the session and penalize the FlowFiles. However, in this case, we can get into some very problematic cases. For example, the Processor may be in a bad state and may continually run, depleting system resources, without providing any useful work. This is fairly common, for instance, when a NullPointerException is thrown continually. In order to avoid this case, if an Exception other than ProcessException is able to escape the Processor’s onTrigger method, the framework will also "Administratively Yield" the Processor. This means that the Processor will not be triggered to run again for some amount of time. The amount of time is configured in the nifi.properties file but is 10 seconds by default.

More often than not, when an Exception occurs in a Processor, it occurs from within a callback (I.e., InputStreamCallback, OutputStreamCallback, or StreamCallback). That is, during the processing of a FlowFile’s content. Callbacks are allowed to throw either RuntimeException or IOException. In the case of RuntimeException, this Exception will propagate back to the onTrigger method. In the case of an IOException, the Exception will be wrapped within a ProcessException and this ProcessException will then be thrown from the Framework.

For this reason, it is recommended that Processors that use callbacks do so within a try/catch block and catch ProcessException as well as any other RuntimeException that they expect their callback to throw. It is not recommended that Processors catch the general Exception or Throwable cases, however. This is discouraged for two reasons.

First, if an unexpected RuntimeException is thrown, it is likely a bug and allowing the framework to rollback the session will ensure no data loss and ensures that DataFlow Managers are able to deal with the data as they see fit by keeping the data queued up in place.

Second, when an IOException is thrown from a callback, there really are two types of IOExceptions: those thrown from Processor code (for example, the data is not in the expected format or a network connection fails), and those that are thrown from the Content Repository (where the FlowFile content is stored). If the latter is the case, the framework will catch this IOException and wrap it into a FlowFileAccessException, which extends RuntimeException. This is done explicitly so that the Exception will escape the onTrigger method and the framework can handle this condition appropriately. Catching the general Exception prevents this from happening.

When an issue occurs during processing, the framework exposes two methods to allow Processor developers to avoid performing unnecessary work: "penalization" and "yielding." These two concepts can become confusing for developers new to the NiFi API. A developer is able to penalize a FlowFile by calling the penalize(FlowFile) method of ProcessSession. This causes the FlowFile itself to be inaccessible to downstream Processors for a period of time. The amount of time that the FlowFile is inaccessible is determined by the DataFlow Manager by setting the "Penalty Duration" setting in the Processor Configuration dialog. The default value is 30 seconds. Typically, this is done when a Processor determines that the data cannot be processed due to environmental reasons that are expected to sort themselves out. A great example of this is the PutSFTP processor, which will penalize a FlowFile if a file already exists on the SFTP server that has the same filename. In this case, the Processor penalizes the FlowFile and routes it to failure. A DataFlow Manager can then route failure back to the same PutSFTP Processor. This way, if a file exists with the same filename, the Processor will not attempt to send the file again for 30 seconds (or whatever period the DFM has configured the Processor to use). In the meantime, it is able to continue to process other FlowFiles.

On the other hand, yielding allows a Processor developer to indicate to the framework that it will not be able to perform any useful function for some period of time. This commonly happens with a Processor that is communicating with a remote resource. If the Processor cannot connect to the remote resource, or if the remote resource is expected to provide data but reports that it has none, the Processor should call yield on the ProcessContext object and then return. By doing this, the Processor is telling the framework that it should not waste resources triggering this Processor to run, because there’s nothing that it can do - it’s better to use those resources to allow other Processors to run.

Thus far, when we have discussed the ProcessSession, we have typically referred to it simply as a mechanism for accessing FlowFiles. However, it provides another very important capability, which is transactionality. All methods that are called on a ProcessSession happen as a transaction. When we decided to end the transaction, we can do so either by calling commit() or by calling rollback(). Typically, this is handled by the AbstractProcessor class: if the onTrigger method throws an Exception, the AbstractProcessor will catch the Exception, call session.rollback(), and then re-throw the Exception. Otherwise, the AbstractProcessor will call commit() on the ProcessSession.

There are times, however, that developers will want to roll back a session explicitly. This can be accomplished at any time by calling the rollback() or rollback(boolean) method. If using the latter, the boolean indicates whether or not those FlowFiles that have been pulled from queues (via the ProcessSession get methods) should be penalized before being added back to their queues.

When rollback is called, any modification that has occurred to the FlowFiles in that session are discarded, to included both content modification and attribute modification. Additionally, all Provenance Events are rolled back (with the exception of any SEND event that was emitted by passing a value of true for the force argument). The FlowFiles that were pulled from the input queues are then transferred back to the input queues (and optionally penalized) so that they can be processed again.

On the other hand, when the commit method is called, the FlowFile’s new state is persisted in the FlowFile Repository, and any Provenance Events that occurred are persisted in the Provenance Repository. The previous content is destroyed (unless another FlowFile references the same piece of content), and the FlowFiles are transferred to the outbound queues so that the next Processors can operate on the data.

It is also important to note how this behavior is affected by using the org.apache.nifi.annotations.behavior.SupportsBatching annotation. If a Processor utilizes this annotation, calls to ProcessSession.commit may not take affect immediately. Rather, these commits may be batched together in order to provide higher throughput. However, if at any point, the Processor rolls back the ProcessSession, all changes since the last call to commit will be discarded and all "batched" commits will take affect. These "batched" commits are not rolled back.

One of the most important concepts to keep in mind when developing a Processor (or any other component) is the user experience that you are creating. It’s important to remember that as the developer of such a component, you may have important knowledge about the context that others do not have. Documentation should always be supplied so that those less familiar with the process are able to use it with ease.

When thinking about the user experience, it is also important to note that consistency is very important. It is best to stick with the standard Naming Conventions. This is true for Processor names, Property names and value, Relationship names, and any other aspect that the user will experience.

Simplicity is crucial! Avoid adding properties that you don’t expect users to understand or change. As developers, we are told that hard-coding values is bad. But this sometimes results in developers exposing properties that, when asked for clarification, tell users to just leave the default value. This leads to confusion and complexity.

For the sake of making a single, cohesive unit, developers are sometimes tempted to combine several functions into a single Processor. This is very true for the case when a Processor expects input data to be in format X so that the Processor can convert the data into format Y and send the newly-formatted data to some external service.

Taking this approach of formatting the data for a particular endpoint and then sending the data to that endpoint within the same Processor has several drawbacks:

In order to avoid these issues, and make Processors more reusable, a Processor should always stick to the principal of "do one thing and do it well." Such a Processor should be broken into two separate Processors: one to convert the data from Format X to Format Y, and another Processor to send data to the remote resource.

In order to deliver a consistent look and feel to users, it is advisable that Processors keep with standard naming conventions. The following is a list of standard conventions that are used:

When creating a Processor, the developer is able to provide hints to the framework about how to utilize the Processor most effectively. This is done by applying annotations to the Processor’s class. The annotations that can be applied to a Processor exist in three sub-packages of org.apache.nifi.annotations. Those in the documentation sub-package are used to provide documentation to the user. Those in the lifecycle sub-package instruct the framework which methods should be called on the Processor in order to respond to the appropriate life-cycle events. Those in the behavior package help the framework understand how to interact with the Processor in terms of scheduling and general behavior.

The following annotations from the org.apache.nifi.annotations.behavior package can be used to modify how the framework will handle your Processor:

An important point to keep in mind is that NiFi provides a generic data processing capability. Data can be in any format. Processors are generally scheduled with several threads. A common mistake that developers new to NiFi make is to buffer all the contents of a FlowFile in memory. While there are cases when this is required, it should be avoided if at all possible, unless it is well-known what format the data is in. For example, a Processor responsible for executing XPath against an XML document will need to load the entire contents of the data into memory. This is generally acceptable, as XML is not expected to be extremely large. However, a Processor that searches for a specific byte sequence may be used to search files that are hundreds of gigabytes or more. Attempting to load this into memory can cause a lot of problems - especially if multiple threads are processing different FlowFiles simultaneously.

Instead of buffering this data into memory, it is advisable to instead evaluate the data as it is streamed from the Content Repository (i.e., scan the content from the InputStream that is provided to your callback by ProcessSession.read). Of course, in this case, we don’t want to read from the Content Repository for each byte, so we would use a BufferedInputStream or somehow buffer some small amount of data, as appropriate.

Just like with the Processor interface, the ControllerService interface exposes methods for configuration, validation, and initialization. These methods are all identical to those of the Processor interface except that the initialize method is passed a ControllerServiceInitializationContext, rather than a ProcessorInitializationContext.

Controller Services come with an additional constraint that Processors do not have. A Controller Service must be comprised of an interface that extends ControllerService. Implementations can then be interacted with only through their interface. A Processor, for instance, will never be given a concrete implementation of a ControllerService and therefore must reference the service only via interfaces that extends ControllerService.

This constraint is in place mainly because a Processor can exist in one NiFi Archive (NAR) while the implementation of the Controller Service that the Processor lives in can exist in a different NAR. This is accomplished by the framework by dynamically implementing the exposed interfaces in such a way that the framework can switch to the appropriate ClassLoader and invoke the desired method on the concrete implementation. However, in order to make this work, the Processor and the Controller Service implementation must share the same definition of the Controller Service interface. Therefore, both of these NARs must depend on the NAR that houses the Controller Service’s interface. See NiFi Archives (NARs) for more information.

ControllerServices may be obtained by a Processor, another ControllerService, or a ReportingTask by means of the ControllerServiceLookup or by using the identifiesControllerService method of the PropertyDescriptor’s Builder class. The ControllerServiceLookup can be obtained by a Processor from the ProcessorInitializationContext that is passed to the initialize method. Likewise, it is obtained by a ControllerService from the ControllerServiceInitializationContext and by a ReportingTask via the ReportingConfiguration object passed to the initialize method.

For most use cases, though, using the identifiesControllerService method of a PropertyDescriptor Builder is preferred and is the least complicated method. In order to use this method, we create a PropertyDescriptor that references a Controller Service as such:

Using this method, the user will be prompted to supply the SSL Context Service that should be used. This is done by providing the user with a drop-down menu from which they are able to choose any of the SSLContextService configurations that have been configured, regardless of the implementation.

In order to make use of this service, the Processor can use code such as:

Note here that SSLContextService is an interface that extends ControllerService. The only implementation at this time is the StandardSSLContextService. However, the Processor developer need not worry about this detail.

Just like with the Processor and ControllerService interfaces, the ReportingTask interface exposes methods for configuration, validation, and initialization. These methods are all identical to those of the Processor and ControllerService interfaces except that the initialize method is passed a ReportingConfiguration object, as opposed to the initialization objects received by the other Components. The ReportingTask also has an onTrigger method that is invoked by the framework to trigger the task to perform its job.

Within the onTrigger method, the ReportingTask is given access to a ReportingContext, from which configuration and information about the NiFi instance can be obtained. The BulletinRepository allows Bulletins to be queried and allows the ReportingTask to submit its own Bulletins, so that information will be rendered to users. The ControllerServiceLookup that is accessible via the Context provides access to ControllerServices that have been configured. However, this method of obtaining Controller Services is not the preferred method. Rather, the preferred method for obtaining a Controller Service is to reference the Controller Service in a PropertyDescriptor, as is discussed in the Interacting with a ControllerService section.

The EventAccess object that is exposed via the ReportingContext provides access to the ProcessGroupStatus, which exposes statistics about the amount of data processed in the past five minutes by Process Groups, Processors, Connections, and other Components. Additionally, the EventAccess object provides access to the ProvenanceEventRecord`s that have been stored in the `ProvenanceEventRepository. These Provenance Events are emitted by Processors when data is received from external sources, emitted to external services, removed from the system, modified, or routed according to some decision that was made.

Each ProvenanceEvent has the ID of the FlowFile, the type of Event, the creation time of the Event, and all FlowFile attributes associated with the FlowFile at the time that the FlowFile was accessed by the component as well as the FlowFile attributes that were associated with the FlowFile as a result of the processing that the event describes. This provides a great deal of information to ReportingTasks, allowing reports to be generated in many different ways to expose metrics and monitoring capabilities needed for any number of operational concerns.

Most unit tests for a Processor or a Controller Service start by creating an instance of the TestRunner class. In order to add the necessary classes to your Processor, you can use the Maven dependency:

We create a new TestRunner by calling the static newTestRunner method of the TestRunners class (located in the org.apache.nifi.util package). This method takes a single argument. That argument can either be the class of the Processor to test or can be an instance of a Processor.

After creating a new Test Runner, we can add any Controller Services to the Test Runner that our Processor will need in order to perform its job. We do this by calling the addControllerService method and supply both an identifier for the Controller Service and an instance of the Controller Service.

If the Controller Service needs to be configured, its properties can be set by calling the setProperty(ControllerService, PropertyDescriptor, String), setProperty(ControllerService, String, String), or setProperty(ControllerService, PropertyDescriptor, AllowableValue) method. Each of these methods returns a ValidationResult. This object can then be inspected to ensure that the property is valid by calling isValid. Annotation data can be set by calling the setAnnotationData(ControllerService, String) method.

We can now ensure that the Controller Service is valid by calling assertValid(ControllerService) - or ensure that the configured values are not valid, if testing the Controller Service itself, by calling assertNotValid(ControllerService).

Once a Controller Service has been added to the Test Runner and configured, it can now be enabled by calling the enableControllerService(ControllerService) method. If the Controller Service is not valid, this method will throw an IllegalStateException. Otherwise, the service is now ready to use.

After configuring any necessary Controller Services, we need to configure our Processor. We can do this by calling the same methods as we do for Controller Services, without specifying any Controller Service. I.e., we can call setProperty(PropertyDescriptor, String), and so on. Each of the setProperty methods again returns a ValidationResult property that can be used to ensure that the property value is valid.

Similarly, we can also call assertValid() and assertNotValid() to ensure that the configuration of the Processor is valid or not, according to our expectations.

Before triggering a Processor to run, it is usually necessary to enqueue FlowFiles for the Processor to process. This can be achieved by using the enqueue methods of the TestRunner class. The enqueue method has several different overrides, and allows data to be added in the form of a byte[], InputStream, or Path. Each of these methods also supports a variation that allows a Map<String, String> to be added to support FlowFile attributes.

Additionally, there is an enqueue method that takes a var-args of FlowFile objects. This can be useful, for example, to obtain the output of a Processor and then feed this to the input of the Processor.

After configuring the Controller Services and enqueuing the necessary FlowFiles, the Processor can be triggered to run by calling the run method of TestRunner. If this method is called without any arguments, it will invoke any method in the Processor with an @OnScheduled annotation, call the Processor’s onTrigger method once, and then run the @OnUnscheduled and finally @OnStopped methods.

If it is desirable to run several iterations of the onTrigger method before the other @OnUnscheduled and @OnStopped life-cycle events are triggered, the run(int) method can be used to specify now many iterations of onTrigger should be called.

There are times when we want to trigger the Processor to run but not trigger the @OnUnscheduled and @OnStopped life-cycle events. This is useful, for instance, to inspect the Processor’s state before these events occur. This can be achieved using the run(int, boolean) and passing false as the second argument. After doing this, though, calling the @OnScheduled life-cycle methods could cause an issue. As a result, we can now run onTrigger again without causing these events to occur by using the run(int,boolean,boolean) version of the run method and passing false as the third argument.

If it is useful to test behavior that occurs with multiple threads, this can also be achieved by calling the setThreadCount method of TestRunner. The default is 1 thread. If using multiple threads, it is important to remember that the run call of TestRunner specifies how many times the Processor should be triggered, not the number of times that the Processor should be triggered per thread. So, if the thread count is set to 2 but run(1) is called, only a single thread will be used.

After a Processor has finished running, a unit test will generally want to validate that the FlowFiles went where they were expected to go. This can be achieved using the TestRunners assertAllFlowFilesTransferred and assertTransferCount methods. The former method takes as arguments a Relationship and an integer to dictate how many FlowFiles should have been transferred to that Relationship. The method will fail the unit test unless this number of FlowFiles were transferred to the given Relationship or if any FlowFile was transferred to any other Relationship. The assertTransferCount method validates only that the FlowFile count was the expected number for the given Relationship.

After validating the counts, we can then obtain the actual output FlowFiles via the getFlowFilesForRelationship method. This method returns a List<MockFlowFile>. It’s important to note that the type of the List is MockFlowFile, rather than the FlowFile interface. This is done because MockFlowFile comes with many methods for validating the contents.

For example, MockFlowFile has methods for asserting that FlowFile Attributes exist (assertAttributeExists), asserting that other attributes are not present (assertAttributeNotExists), or that Attributes have the correct value (assertAttributeEquals, assertAttributeNotEquals). Similar methods exist for verifying the contents of the FlowFile. The contents of a FlowFile can be compared to a byte[], and InputStream, a file, or a String. If the data is expected to be textual, the String version is preferred, as it provides a more intuitive error message if the output is not as expected.

One of the biggest problems when testing a NiFi processor that connects to a remote resource is that we don’t want to actually connect to some remote resource from a unit test. We can stand up a simple server ourselves in the unit test and configure the Processor to communicate with it, but then we have to understand and implement the server-specific specification and may not be able to properly send back error messages, etc. that we would like for testing.

Generally, the approach taken here is to have a method in the Processor that is responsible for obtaining a connection or a client to a remote resource. We generally mark this method as protected. In the unit test, instead of creating the TestRunner by calling TestRunners.newTestRunner(Class) and providing the Processor class, we instead create a subclass of the Processor in our unit test and use this:

This allows us to implement a Client that mocks out all of the network communications and returns the different error results that we want to test, as well as ensure that our logic is correct for handling successful calls to the client.

In addition to the above-mentioned capabilities provided by the testing framework, the TestRunner provides several convenience methods for verifying the behavior of a Processor. Methods are provided for ensuring that the Processor’s Input Queue has been emptied. Unit Tests are able to obtain the ProcessContext, ProcessSessionFactory, ProvenanceReporter, and other framework-specific entities that will be used by the TestRunner. The shutdown method provides the ability to test Processor methods that are annotated to be run only on shutdown of NiFi. Annotation Data can be set for Processors that make use of Custom User Interfaces. Finally, the number of threads that should be used to run the Processor can be set via the setThreadCount(int) method.

The back end of Apache NiFi is written in Java. The web tier makes use of JAX-RS and JavaScript is extensively used to provide a user interface. We depend on several third-party JavaScript libraries, including D3 and JQuery, among others. We make use of Apache Maven for our builds and Git for our version control system.

NiFi’s Jira page can be used to find tickets that are tagged as "beginner", or you can dig into any of the tickets for creating Processors. Processors should be self-contained and not rely on other outside components (except for Controller Services), so they make for excellent starting points for new NiFi developers to get started. This exposes the developer to the NiFi API and is the most extensible part of the dataflow system.

Contributions can be provided either by creating a patch:

git format-patch

and attaching that patch to a ticket, or by generating a Pull Request.

The developer mailing list (dev@nifi.incubator.apache.org) is monitored pretty closely, and we tend to respond pretty quickly. If you have a question, don’t hesitate to shoot us an e-mail - we’re here to help! Unfortunately, though, e-mails can get lost in the shuffle, so if you do send an e-mail and don’t get a response within a day or two, it’s our fault - don’t worry about bothering us. Just ping the mailing list again.