NiFi中Groovy调用存储过程报MissingMethodException如何解决

API 探路者

2026-8-29

问题描述

在Apache NiFi中使用Groovy脚本调用存储过程，参考Groovy SQL标准调用方式编写代码，官方参考示例如下：

sql.call("{call CONCAT_NAME(?, ?, ?)}", [Sql.VARCHAR, 'Dierk', 'Koenig']) {
 fullname -> assert fullname == 'Dierk Koenig'
}

脚本实现逻辑：

导入依赖包，从NiFi会话获取flowFile，无flowFile时直接返回
读取flowFile的TYPE_NOTIFICATION、ID_NOTIFICATION两个属性
定义调用存储过程PKG_TEST.P_TEST的SQL语句，组装含入参、输出参数JDBC类型的参数列表
通过名为mydbxx的SQL控制器服务执行call方法，在闭包中接收存储过程返回的输出参数写入flowFile属性
将flowFile路由到REL_SUCCESS成功关系

编写的完整脚本代码如下：

import org.apache.commons.io.IOUtils
import org.apache.nifi.controller.ControllerService
import org.apache.nifi.processor.io.StreamCallback
import java.nio.charset.*
import groovy.sql.OutParameter
import groovy.sql.Sql
import java.sql.ResultSet


////Get the session values from Nifi flow Start 
def flowFile = session.get()
if(!flowFile) return
  String TYPE_NOTIFICATION = flowFile.getAttribute('TYPE_NOTIFICATION')
  String ID_NOTIFICATION = flowFile.getAttribute('ID_NOTIFICATION')
////Get the session values from Nifi flow End
 
 
String sqlString ="""{call PKG_TEST.P_TEST(?,?,?,?,?,?,?,?,?,?,?)}""";

def parametersList = [ID_NOTIFICATION, TYPE_NOTIFICATION,Sql.VARCHAR,Sql.VARCHAR,Sql.DATE,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.DATE ];
 
 
SQL.mydbxx.call(sqlString, parametersList) {out1, out2,...->
    flowFile.putAttribute("out1",out1)...
};

 
session.transfer(flowFile, REL_SUCCESS)

脚本运行抛出如下异常：

2022-06-17 13:38:53,353 ERROR [Timer-Driven Process Thread-9]
o.a.n.p.groovyx.ExecuteGroovyScript
ExecuteGroovyScript[id=26ab18f1-3b0c-18cf-d90b-3d5904676458]
groovy.lang.MissingMethodException: No signature of method:
Script6a6d0a35$_run_closure1.doCall() is applicable for argument
types: (String, String, java.sql.Date, null, String, null, null,
String...) values: [xxxx, xxxx, 2022-05-30, null, OK, null, null,
...]: groovy.lang.MissingMethodException: No signature of method:
Script6a6d0a35$_run_closure1.doCall() is applicable for argument
types: (String, String, java.sql.Date, null, String, null, null,
String...) values: [xxxx,xxxx, 2022-05-30, null, OK, null, null, ...]

异常提示闭包doCall方法签名与实际传入的参数类型（String、java.sql.Date、null等）不匹配。

故障原因

异常由三个核心问题导致：

闭包形参使用了无效的...占位写法，不属于合法Groovy语法。当前存储过程共11个占位符，前2位是入参，后9位是输出参数，call方法执行完成后会按顺序将9个输出值传入闭包，但闭包仅声明了2个形参加无效占位符，形参数量、签名和实际传入的9个参数完全不匹配，直接触发MissingMethodException。
代码中flowFile.putAttribute("out1",out1)...同样是占位写法，未实现实际的属性赋值逻辑；且NiFi flowFile为不可变对象，调用putAttribute会返回新的对象引用，不重新赋值的情况下属性写入不会生效。
NiFi flowFile属性不支持null值，即便参数匹配成功，遇到存储过程返回null的输出参数时也会抛出运行时异常。

修复方案

按以下步骤调整代码即可解决问题：

按输出参数的顺序、数量，在闭包中显式声明全部9个形参，删除无效的...占位
逐个写入属性时，重新接收putAttribute返回的新flowFile引用，同时对null值做转空字符串处理，避免NiFi属性赋值报错
建议将SQL执行逻辑放在try-catch块中，捕获数据库调用异常时将flowFile路由到失败关系，避免流程静默失败

修复后的可运行代码示例：

import groovy.sql.Sql

// 从NiFi会话获取flowFile
def flowFile = session.get()
if(!flowFile) return

String TYPE_NOTIFICATION = flowFile.getAttribute('TYPE_NOTIFICATION')
String ID_NOTIFICATION = flowFile.getAttribute('ID_NOTIFICATION')

String sqlString = "{call PKG_TEST.P_TEST(?,?,?,?,?,?,?,?,?,?,?)}"
// 参数顺序严格和存储过程定义顺序一致：前2位为入参值，后9位为OUT参数类型
def parametersList = [
    ID_NOTIFICATION, 
    TYPE_NOTIFICATION,
    Sql.VARCHAR, Sql.VARCHAR, Sql.DATE,
    Sql.VARCHAR, Sql.VARCHAR, Sql.VARCHAR,
    Sql.VARCHAR, Sql.VARCHAR, Sql.DATE
]

try {
    SQL.mydbxx.call(sqlString, parametersList) {
        out1, out2, out3, out4, out5, out6, out7, out8, out9 ->
        // 写入属性时null转空串，接收新的flowFile引用
        flowFile = flowFile.putAttribute("out1", out1?.toString() ?: "")
        flowFile = flowFile.putAttribute("out2", out2?.toString() ?: "")
        flowFile = flowFile.putAttribute("out3", out3?.toString() ?: "")
        flowFile = flowFile.putAttribute("out4", out4?.toString() ?: "")
        flowFile = flowFile.putAttribute("out5", out5?.toString() ?: "")
        flowFile = flowFile.putAttribute("out6", out6?.toString() ?: "")
        flowFile = flowFile.putAttribute("out7", out7?.toString() ?: "")
        flowFile = flowFile.putAttribute("out8", out8?.toString() ?: "")
        flowFile = flowFile.putAttribute("out9", out9?.toString() ?: "")
    }
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    log.error("调用存储过程失败", e)
    flowFile = flowFile.putAttribute("error_msg", e.getMessage() ?: "")
    session.transfer(flowFile, REL_FAILURE)
}

注意：如果存储过程的参数顺序、IN/OUT模式和上述假设不一致，需要对应调整parametersList的入参值、OUT类型顺序，以及闭包的形参个数，保证三者完全对应。

内容的提问来源于stack exchange，提问作者KnowledgeSeeker