apache-nifi - 流文件卡在队列中（Apache NiFi）

Question

我有以下流程：

ListFTP -> RouteOnAttribute -> FetchFTP -> UnpackContent -> ExecuteScript.

一些文件卡在队列中UnpackContent -> ExecuteScript。

ExecuteScript吃了一些流文件，它们就消失了：failure关系success是空的。它只是显示了一些Tasks/Time现场活动。他们之前都排在队列中ExecuteScript。我试图清空队列，但并非所有流文件都已从此队列中删除。其中大约 1/3 仍然排在队列中。我试图再次禁用所有处理器并清空队列，但它返回：0 FlowFiles (0 bytes) were removed from the queue.

当我尝试更改连接目标时，它返回：

Cannot change destination of Connection because FlowFiles from this Connection are currently held by ExecuteScript[id=d33c9b73-0177-1000-5151-83b7b938de39]

ExecuScript从这个答案（使用Python）。

所以，我不能清空队列，因为它总是返回没有任何流文件的消息，我不能删除连接。这已经持续了几个小时。

连接配置：

调度设置为 0 秒，对流文件等没有惩罚。

是脚本问题吗？

更新

将脚本更改为：

flowFile = session.get() 
if (flowFile != None):
    # All processing code starts at this indent
    if errorOccurred:
        session.transfer(flowFile, REL_FAILURE)
    else:
        session.transfer(flowFile, REL_SUCCESS)
# implicit return at the end

结果相同。

更新 v2

我将并发任务设置为 50，然后ExecuteScript再次运行并终止它。我收到了这个错误：

更新 v3

我用相同的脚本创建了额外的 ExecuteScript 处理器，它工作正常。但是在我停止这个新处理器并创建新的流文件之后，这个处理器现在有同样的问题：它只是卡住了。

好笑。是ExecuteScript一次性使用吗？

标签： apache-nifi

Answer 1

您需要修改 nifi-1.13.2 中的代码，因为 NIFI-8080 导致了这些错误。或者你只使用 nifi 1.12.1

JythonScriptEngineConfigurator：

@Override
public Object init(ScriptEngine engine, String scriptBody, String[] modulePaths) throws ScriptException {
    // Always compile when first run
    if (engine != null) {
            // Add prefix for import sys and all jython modules
            prefix = "import sys\n"
                    + Arrays.stream(modulePaths).map((modulePath) -> "sys.path.append(" + PyString.encode_UnicodeEscape(modulePath, true) + ")")
                    .collect(Collectors.joining("\n"));
    }
    return null;
}

@Override
public Object eval(ScriptEngine engine, String scriptBody, String[] modulePaths) throws ScriptException {
    Object returnValue = null;
    if (engine != null) {
        returnValue = ((Compilable) engine).compile(prefix + scriptBody).eval();
    }
    return returnValue;
}