Hi All,
I am facing issues with Execute Script processor. I am executing Groovy script which have oracle DB in Execute Script processor but its not closing the DB connection.
Issues: Database connection are not closing and its making Inactive connection in database.
The Groovy script we added proper open and close connections. Could you help fix this issues, also let me know where I am not correct .
DB Connection Configuration is like that :
Execute Script Process are configured as fallowing:
Groovy Script as following :
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 oracle.jdbc.OracleTypes
import java.sql.ResultSet
def lookup = context.controllerServiceLookup
def dbServiceName = ConnectionPoolScriptExecution.value
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection();
def sql = new Sql(conn);
def flowFile = session.get()
if(!flowFile) return
//Session Values
strApplication = flowFile.getAttribute('application')
strUsername = flowFile.getAttribute('username')
strPassword = flowFile.getAttribute('password')
// SQL Query Creation and execution
def ValidOrNot
String sqlString = "select FN_HP_IS_VALID_APP('${strApplication}','${strUsername}','${strPassword}') as ValidOrNot from dual";
try {
rowNum = 0
sql.eachRow(sqlString, { dataResult ->
//println(dataResult.ValidOrNot)
ValidOrNot = dataResult.ValidOrNot
})
}
catch(Exception ex){
conn?.close()
}
finally {
conn?.close()
}
conn?.close()
// Assign values to session object
def attrMap = ['VALIDORNOT':ValidOrNot,'sqlString':sqlString]
flowFile = session.putAllAttributes(flowFile, attrMap)
session.transfer(flowFile, REL_SUCCESS)
Hello everyone!
In the company where I work they had the same problem, this time I had some time to review the script and do a proof of concept to try to solve the problem.
In my test, after a few flowfiles the processor generates the oracle database connection error.
Searching groovy documentation I found that the example that is floating around the internet is not correctly using the closure for the connection (this incident for example).
The solution is to close the instance along with the connection, currently it is trying to close the connection that is declared at the beginning of the script but the instance is ignored. In my tests, when closing the connection instance, the process executed 30,000 flowfiles without generating any problems. Instead, for the definition of this script, after 14,000 flowfiles, the processor started to generate a connection error.
You can find more info in the groovy documentancion:
Connecting to HSQLDB
Well, what was expected arrived... the solution is as follows:
You need to add the finally statement and close sql
//Imports
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 conncation string from NIFI conncation pool
def lookup = context.controllerServiceLookup
//Este valor se crea en el procesador con el nombre del controlador de la conexion
def dbServiceName = databaseConnectionPoolName.value
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
sql = Sql.newInstance(conn);
////Get the session values from Nifi flow Start
try {
flowFile = session.get()
if(!flowFile) return
PRIMER_PARAMETRO = flowFile.getAttribute('PRIMER_PARAMETRO')
SEGUNDO_PARAMETRO = flowFile.getAttribute('SEGUNDO_PARAMETRO')
// Procedimiento con 2 parametros de entrada y 2 de salida
String sqlString ="""{call PRUEBA_CONCEPTO.PROC_DUMMY_DUAL(?, ?, ?, ?)}""";
//Parametros de salida
def CODIGO
def MENSAJE
// Lista de parametros del procedimiento
def parametersList = [PRIMER_PARAMETRO, SEGUNDO_PARAMETRO, Sql.NUMERIC, Sql.VARCHAR];
//Ejecucion del procedimiento
sql.call(sqlString, parametersList) {out_status_code, out_status_desc ->
CODIGO = out_status_desc
MENSAJE = out_status_code
};
//Set the session values start
def attrMap = ['status_desc':CODIGO, 'status_code':String.valueOf(MENSAJE)]
flowFile = session.putAllAttributes(flowFile, attrMap)
session.transfer(flowFile, REL_SUCCESS)
} catch (e){
log.error('Scripting error', e)
flowFile = session.putAttribute(flowFile, "error", e.getMessage())
session.transfer(flowFile, REL_FAILURE)
} finally {
if (conn != null) conn.close();
if (sql != null) sql.close();
}
In my case I use SQL.instance with a DBCPConnectionPool but it should be the same for this case.
Solution
Issue
Dummy procedure:
CREATE OR REPLACE
PACKAGE PRUEBA_CONCEPTO AS
/* TODO enter package declarations (types, exceptions, methods etc) here */
PROCEDURE PROC_DUMMY_DUAL ( PRIMER_PARAMETRO IN VARCHAR2
, SEGUNDO_PARAMETRO IN NUMBER
, CODIGO OUT NUMBER
, MENSAJE OUT VARCHAR2
);
END PRUEBA_CONCEPTO;
/
CREATE OR REPLACE
PACKAGE BODY PRUEBA_CONCEPTO AS
PROCEDURE PROC_DUMMY_DUAL ( PRIMER_PARAMETRO IN VARCHAR2
, SEGUNDO_PARAMETRO IN NUMBER
, CODIGO OUT NUMBER
, MENSAJE OUT VARCHAR2
) AS
BEGIN
-- TAREA: Se necesita implantación para PROCEDURE PRUEBA_CONCEPTO.PROC_DUMMY_DUAL
CODIGO := SEGUNDO_PARAMETRO;
MENSAJE := PRIMER_PARAMETRO;
END PROC_DUMMY_DUAL;
END PRUEBA_CONCEPTO;
Test Procedure
SET SERVEROUTPUT ON;
DECLARE
PRIMER_PARAMETRO VARCHAR2(100);
SEGUNDO_PARAMETRO NUMBER;
CODIGO NUMBER;
MENSAJE VARCHAR2(100);
BEGIN
PRIMER_PARAMETRO := 'eSTO ES UN MENSAJE';
SEGUNDO_PARAMETRO := 1;
PRUEBA_CONCEPTO.PROC_DUMMY_DUAL(PRIMER_PARAMETRO, SEGUNDO_PARAMETRO, CODIGO, MENSAJE);
DBMS_OUTPUT.PUT_LINE('CODIGO : ' || CODIGO);
DBMS_OUTPUT.PUT_LINE('MENSAJE : ' || MENSAJE);
END;
