NiFi: Send flowFile @OnflowFile

Question

Is it possible to send flowfile on @OnStopped annotation?

Basically, I want to write custom processor which can send an attribute in flowFile on the stop of processor.

Any suggestions?

I was trying below :

ProcessSession session;

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
    FlowFile flowFile = session.get();
    if (flowFile == null) {
        flowFile = session.create();
    }
    flowFile = session.putAttribute(flowFile, "ATTRIBUTE_SIGNAL", "start");
    session.transfer(flowFile, success);
}

@OnStopped
public void sendStop() {
    FlowFile flowFile = session.get();
    flowFile = session.create();
    flowFile = session.putAttribute(flowFile, "ATTRIBUTE_SIGNAL", "stop");
    session.transfer(flowFile, success);
}

but it is failing with

2018-04-30 20:44:25,540 ERROR [StandardProcessScheduler Thread-3] org.apache.nifi.util.ReflectionUtils Failed while invoking annotated method 'public void com.kotak.nifi.processors.streaming.SignalGenerator.sendStop()' with arguments '[]'.
java.lang.reflect.InvocationTargetException: null.

Bryan Bende Bryan Bende · Accepted Answer · 2018-04-30T14:08:01

The lifecycle methods like OnScheduled/OnStopped/etc are not really meant to produce flow files which is why you don't have access to a ProcessSession, only onTrigger does.

Processors are generally supposed to be loosely coupled where one processor doesn't really know/care about other processors, it just takes a flow files out of a queue and processes them.

Technically you may be able to achieve what you want by storing a reference to a ProcessSession that you got in onTrigger in a member variable of the processor so you can use it later in OnStopped.

NiFi: Send flowFile @OnflowFile

1 Answers