Custom Processor Development Workflow

Custom processor development in Apache NiFi follows a structured workflow: code implementation following NiFi conventions → packaging as NAR file → deployment and usage.

Implementation Following NiFi Conventions

NiFi provides extensible processor interfaces and abstract classes. Developers implement custom processors by extending these classes and overrriding required methods.

NAR Packaging

NiFi processors are distributed as NAR (NiFi Archive) files. Custom code must be packaged according to specific rules to generate valid NAR files.

Deployment and Usage

Completed NAR file are placed in the lib directory of the NiFi installation, alongside built-in processor NAR files.

Project Setup Approaches

Two primary approaches exist for setting up custom processor projects:

• Official Bundle Structure - Following NiFi's recommended multi-module Maven project layout
• Standalone Maven Project - Single-module approach with custom NAR packaging configuration

Both approaches require these fundamental steps:

1. Create META-INF/services directory in resources
2. Create org.apache.nifi.processor.Processor file containing fully-qualified processor class names
3. Configure project for NAR packaging

Official Bundle Structure

This approach uses NiFi's recommended multi-module Maven structure:

• Root project (pom packaging) managing submodules
• Processors module (jar packaging) containing implementation code
• NAR module (nar packaging) handling final packaging

Example root POM structure:

<project>
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.apache.nifi</groupId>
        <artifactId>nifi-external</artifactId>
        <version>1.15.0</version>
    </parent>
    
    <artifactId>custom-processor-bundle</artifactId>
    <packaging>pom</packaging>
    
    <modules>
        <module>custom-processors</module>
        <module>custom-processor-nar</module>
    </modules>
</project>

Processors module POM includes necessary dependencies:

<dependencies>
    <dependency>
        <groupId>org.apache.nifi</groupId>
        <artifactId>nifi-api</artifactId>
        <version>1.15.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.nifi</groupId>
        <artifactId>nifi-processor-utils</artifactId>
        <version>1.15.0</version>
    </dependency>
</dependencies>

Processor implementation example:

@Tags({"example", "resource"})
@CapabilityDescription("Loads resource from NAR and writes to FlowFile content")
public class ResourceStreamWriter extends AbstractProcessor {
    
    public static final Relationship SUCCESS = new Relationship.Builder()
        .name("success")
        .description("Successfully processed files").build();
    
    private Set<Relationship> relationships;
    private String resourceContent;
    
    @Override
    protected void init(ProcessorInitializationContext context) {
        Set<Relationship> rels = new HashSet<>();
        rels.add(SUCCESS);
        this.relationships = Collections.unmodifiableSet(rels);
        
        InputStream resourceStream = getClass()
            .getClassLoader().getResourceAsStream("data.txt");
        try {
            this.resourceContent = IOUtils.toString(resourceStream, 
                StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new RuntimeException("Resource loading failed", e);
        }
    }
    
    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }
    
    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) {
        FlowFile flowFile = session.get();
        if (flowFile == null) return;
        
        flowFile = session.write(flowFile, out -> {
            out.write(resourceContent.getBytes(StandardCharsets.UTF_8));
        });
        session.transfer(flowFile, SUCCESS);
    }
}

Standalone Maven Project

Simpler approach using single-module Mavenn project with custom NAR configuration:

<project>
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example.nifi</groupId>
    <artifactId>custom-processor</artifactId>
    <version>1.0.0</version>
    <packaging>nar</packaging>
    
    <properties>
        <nifi.version>1.15.0</nifi.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.nifi</groupId>
            <artifactId>nifi-api</artifactId>
            <version>${nifi.version}</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.nifi</groupId>
                <artifactId>nifi-nar-maven-plugin</artifactId>
                <version>1.3.1</version>
                <extensions>true</extensions>
            </plugin>
        </plugins>
    </build>
</project>

Unit Testing Processors

NiFi provides testing utilities through the nifi-mock dependency:

<dependency>
    <groupId>org.apache.nifi</groupId>
    <artifactId>nifi-mock</artifactId>
    <version>${nifi.version}</version>
    <scope>test</scope>
</dependency>

TestRunner Usage

public class CustomProcessorTest {
    private TestRunner runner = TestRunners.newTestRunner(CustomProcessor.class);
    
    @Test
    public void testProcessorExecution() {
        runner.setProperty("threshold", "100");
        runner.assertValid();
        
        runner.enqueue("test data content");
        runner.run();
        
        runner.assertTransferCount("success", 1);
        List<MockFlowFile> results = runner.getFlowFilesForRelationship("success");
        
        MockFlowFile result = results.get(0);
        result.assertContentEquals("expected content");
    }
}

Key Processor Concepts

Core Components

• FlowFile: Data unit with associated attributes
• ProcessSession: Interface for FlowFile operations
• ProcessContext: Provides processor configuration access
• PropertyDescriptor: Defines configurable properties
• Relationship: Defines FlowFile routing destinations

AbstractProcessor Lifecycle

Key methods to override:

• init(): Processor initialization
• getRelationships(): Define output relationships
• getSupportedPropertyDescriptors(): Configure processor properties
• onTrigger(): Main processing logic execution

Example property definition:

public static final PropertyDescriptor THRESHOLD = new PropertyDescriptor.Builder()
    .name("Threshold Value")
    .description("Processing threshold")
    .required(true)
    .addValidator(StandardValidators.INTEGER_VALIDATOR)
    .build();

Processor Execution

Processors are triggered when:

• Incoming connections contain FlowFiles
• No incoming connections exist
• Annotated with @TriggerWhenEmpty

Threading considerations:

• @TriggerSerially ensures single-threaded execution
• Default behavior allows concurrent execution
• Thread safety must be considered for shared resources