Hello if you could help me with this error. Since I do not understand the error. Attached image. Regards My code: import org.apache.nifi.controller.ControllerService
import groovy.sql.Sql
def lookup = context.controllerServiceLookup
def dbServiceName = databaseConnectionPoolName.value
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
try {
def flowFileIn = session.get()
log.info("flow in : "+flowFileIn.getAttribute('Tablas'))
def tables = flowFileIn.getAttribute('Tablas')
def rutas = flowFileIn.getAttribute('Ruta')
flowFile = session.create()
def schemaName = schema.evaluateAttributeExpressions(flowFile).value
//def tables = ruta.evaluateAttributeExpressions(flowFile).value
//def rutas = tablas.evaluateAttributeExpressions(flowFile).value
log.info("schemaName property value: "+schemaName)
log.info("tablas in : "+tables)
log.info("rutas in : "+rutas)
def sql = new Sql(conn)
def tablesArray = tables.trim().split(',');
def rutasArray = rutas.trim().split(',');
log.info("Table atrribute value: "+tablesArray)
log.info("rutas atrribute value: "+rutasArray)
for ( file in rutasArray ) {
sql.execute("begin dbms_logmnr.add_logfile( logfilename =>'"+file+"'); end;")
//log.info("archivo a buscar: "+file)
}
sql.execute(' begin dbms_logmnr.start_logmnr(options => dbms_logmnr.dict_from_online_catalog); end;')
for ( table in tablesArray ) {
def countRow = 0
flowFile = session.write(flowFile, {out ->
query = "select sql_redo from v\$logmnr_contents where OPERATION = 'INSERT' and SEG_OWNER = '"+schemaName+"' and TABLE_NAME = '"+table+"' "
log.info("query a lanzar: "+query)
sql.rows("select sql_redo from v\$logmnr_contents where OPERATION = 'INSERT' and SEG_OWNER = '"+schemaName+"' and TABLE_NAME = '"+table+"' ").eachWithIndex { row, idx ->
if(idx == 0) { out.write(((row.keySet() as List).join(',') + "\n").getBytes()) }
out.write((row.values().join(',') + "\n").getBytes())
countRow = countRow + 1
}
log.info("cantidad"+countRow+" para la tabla: "+table)
} as OutputStreamCallback)
if (countRow >= 1){
//flowFile = session.putAttribute(flowFile, 'table', table)
session.transfer(flowFile, REL_SUCCESS)
} else {
conn?.close()
log.info('No hay datos para la tabla: '+table)
session.transfer(flowFile, REL_FAILURE)
}
}
sql.execute(' begin dbms_logmnr.end_logmnr(); end;')
} catch(e) {
conn?.close()
log.error('Scripting error', e)
session.transfer(flowFile, REL_FAILURE)
}
conn?.close()
... View more
