java - NiFi : Send flowFile @OnflowFile

Question

是否可以在 @OnStopped 注释上发送流文件？

基本上，我想编写自定义处理器，它可以在处理器停止时在 flowFile 中发送一个属性。

有什么建议吗？

我在下面尝试:

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);
}

但是失败了

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.

Answer 1

像 OnScheduled/OnStopped/等这样的生命周期方法并不真正意味着生成流文件，这就是为什么你不能访问 ProcessSession，只有 onTrigger 可以。

处理器通常应该是松散耦合的，其中一个处理器并不真正了解/关心其他处理器，它只是从队列中取出一个流文件并处理它们。

从技术上讲，您可以通过在处理器的成员变量中存储对在 onTrigger 中获得的 ProcessSession 的引用来实现您想要的效果，以便稍后在 OnStopped 中使用它。