groovy - 访问NiFi流文件的沿袭

我正在为NiFi的流文件开发某种错误处理，例如数据库子系统拒绝从流文件写入数据，因为该数据与预期不符，因为该数据的源系统缺少一些主数据。
因此，此错误处理将数据写入MongoDB，并提供更多错误信息。
这些“更多信息”之一是此流文件的某种堆栈跟踪，即数据沿袭。为此，我编写了一个具有Groovy脚本的InvokeScriptedProcessor来实现此目的。

这是脚本的重要部分：

ArrayList getStacktrace(flowfileUuid){
    def lineage = this.provenanceRepository.createLineageQuery(flowfileUuid)
    def lineageData = this.provenanceRepository.getLineageData(lineage.id)

    if (lineageData.results == null || lineageData.results.nodes.size() == 0){
        println "cannot find stacktrace for ${flowfileUuid}."
        return []
    }
    def eventIds = lineageData.results.nodes.findAll {n -> n.type == 'EVENT'}.collect {n -> n.id }.sort()
    def provenanceEvents = []
    for (eventId in eventIds){
        provenanceEvents << this.provenanceRepository.getProvenanceEvent(eventId).provenanceEvent.componentName
    }
    this.provenanceRepository.deleteLineageQuery(lineage.id)
    return provenanceEvents
}

对于createLineageQuery，我将POSTING添加到nifi-api中，并在其中将/nifi-api/provenance/lineage添加流文件的uuid。结果除其他外是查询的ID。我正在使用ID到getLineageData;还有一个属性finished，我正在等待查询完成。
使用此沿袭数据，我getProvenanceEvent数据并将组件（处理器）的名称写入数组。
之后，按照文档中的说明deleteLineageQuery。

所以这就是我的堆栈跟踪。

现在的问题是，当流文件首次命中此InvokeScriptedProcessor时，沿袭数据为空。我尝试了很多事情，例如等待和尝试。无济于事。
现在奇怪的是，当我重播此处理器的流文件时，沿袭数据不为空。
因此，行为并不像我期望的那样是确定性的。
有时，当我第一次处理流文件时，谱系数据不为空。
我也曾与Fiddler一起尝试过这种东西，并且一直都在工作。

我的方法有问题吗？
我当前正在使用NiFi 1.6.0。

编辑：
我将以布莱恩的答案作为解决方案。
如果有时间，我会立即调查，但听起来是正确的。但是，我尝试使用NiFi 1.8.0解决方案，它可以按预期工作。因此，目前我对第一步的实现方式感到满意，但我会根据Bryan的建议来改进解决方案。

最佳答案

我不能完全确定问题出在哪里，但是总的来说，起源数据并不是真正要从处理器访问的，这就是为什么会话或上下文没有提供API来让您检索起源事件，而仅创建事件的原因被允许。

为了运行出处查询，需要对事件进行索引，并且无法保证何时进行索引与处理流文件的时间有关。因此，事件可能还不可见。

ReportingTask是访问出处事件的预期方式，可用于将其从NiFi推送到某个外部系统以进行长期存储。

关于groovy - 访问NiFi流文件的沿袭，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/53534318/