One-and-done Flows (aka Job Processing)

We get a lot of questions on the Apache NiFi mailing list about being able to run a flow/processor just once, and indeed there is a related feature request Jira (NIFI-1407) that remains unimplemented. IMO this is for a few reasons, one being that "one-shot" processing doesn't really align with a flow-based architecture such as NiFi's, it is really more of a job processing feature.

Having said that, many users still desire/need that capability, and some workarounds have been presented, such as scheduling the source processor (often GenerateFlowFile if used as a trigger) for an exorbitant amount of time (weeks, e.g.) knowing that probably someone will come in and restart the processor before that. The other is to start the processor from the UI and immediately stop it, that keeps the original run going but prevents the processor from being scheduled further. The same can be done from the REST API, although for secure NiFi instances/clusters that can be a bit painful to set up.

In that vein I wanted to tackle one common use case in this area, the one where a flow file needs to be generated as a trigger, with or without possible content. I'll present two different Groovy scripts to be used in an ExecuteScript processor, one to use downstream from GenerateFlowFile, and one as the source processor itself.

For the first scenario, I'm picturing a GenerateFlowFile as the source processor, although technically it could be any source processor. The key is to schedule it to run at some huge interval, like 1 year or something. This is normally a decent workaround, but technically doesn't work if you leave your flow running for a year. However having this long an interval will prevent a ton of provenance events for the flow files headed into the ExecuteScript.

This Groovy script uses the processor's StateManager to keep track of a boolean variable "one_shot". If it is not yet set, the flow file is transferred to success and then the variable "one_shot" is set. From then on, the flow files will just be removed from the session. Note that this creates a DROP provenance event, which is why we don't want a lot of CREATE/DROP events for this part of the flow, which reinforces my earlier point about scheduling the source processor.  The script is as follows:

import org.apache.nifi.components.state.*

flowFile = session.get()
if(!flowFile) return
StateManager sm = context.stateManager
def oldState = new HashMap<String,String>()
oldState.putAll(sm.getState(Scope.LOCAL).toMap())
if(!oldState.one_shot) {
  // Haven't sent the flow file yet, do it now and update state
  session.transfer(flowFile, REL_SUCCESS)
  oldState.one_shot = 'true'
  sm.setState(oldState, Scope.LOCAL)
} else {
  // Remove flow file -- NOTE this causes upstream data loss, only use
  // when the upstream flow files are meant as triggers. Also be sure the
  // upstream schedule is large like 1 day, to avoid unnecessary provenance events.
  session.remove(flowFile)
}
 
An alternative to dropping the flow files is to rollback the session, that can prevent data loss but eventually will lead to backpressure being applied at the source processor. If that makes more sense for your use case, then just replace the session.remove() with session.rollback().

The other scenario is when you don't need a full-fledged GenerateFlowFile processor (or any other processor), then ExecuteScript can be the source processor. The following script works similarly to the previous script, except it generates rather than receives flow files, and will accept an optional user-defined property called "content", which it will use for the outgoing FlowFile content. Again the run schedule should be sufficiently large, but in this case it won't matter as much since no provenance events will be created. However a thread is created and destroyed at each scheduled run, so might as well set the run schedule to be a large value. The script is as follows:

import org.apache.nifi.components.state.*

StateManager sm = context.stateManager
def oldState = new HashMap<String,String>()
oldState.putAll(sm.getState(Scope.LOCAL).toMap())
if(!oldState.one_shot) {
  // Haven't sent the flow file yet, do it now and update state
  def flowFile = session.create()
  try {
    String ffContent = context.getProperty('content')?.evaluateAttributeExpressions()?.value
    if(ffContent) {
      flowFile = session.write(flowFile, {outputStream -> 
        outputStream.write(ffContent.bytes)
      } as OutputStreamCallback)
    }
    session.transfer(flowFile, REL_SUCCESS)
  } catch(e) {
    log.error("Couldn't generate FlowFile", e)
    session.remove(flowFile)
  }
  oldState.one_shot = 'true'
  sm.setState(oldState, Scope.LOCAL)
}

To reset these processors (in order to run another one-shot job/flow), you can stop them, right-click and choose View State then Clear (or send the corresponding command(s) via the REST API).

As always, please let me know how/if this works for you. Cheers!