This is an automated email from the ASF dual-hosted git repository. joewitt pushed a commit to branch support/nifi-1.13 in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 2e58491e6dd9e3029b62fb8a30097df169a2aa97 Author: Matthew Burgess <[email protected]> AuthorDate: Tue Feb 23 18:18:42 2021 -0500 NIFI-8249: Fixed issue with error during multiple FlowFile results in ExecuteSQL processors --- .../processors/standard/AbstractExecuteSQL.java | 3 +- .../processors/standard/TestExecuteSQLRecord.java | 43 ++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java index 33a3dbb..51f953a 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java @@ -357,8 +357,9 @@ public abstract class AbstractExecuteSQL extends AbstractProcessor { fragmentIndex++; } catch (Exception e) { - // Remove the result set flow file and propagate the exception + // Remove any result set flow file(s) and propagate the exception session.remove(resultSetFF); + session.remove(resultSetFlowFiles); if (e instanceof ProcessException) { throw (ProcessException) e; } else { diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteSQLRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteSQLRecord.java index 723f141..fe17e94 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteSQLRecord.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteSQLRecord.java @@ -277,6 +277,49 @@ public class TestExecuteSQLRecord { } @Test + public void testWithOutputBatchingLastBatchFails() throws InitializationException, SQLException { + // remove previous test database, if any + final File dbLocation = new File(DB_LOCATION); + dbLocation.delete(); + + // load test data to database + final Connection con = ((DBCPService) runner.getControllerService("dbcp")).getConnection(); + Statement stmt = con.createStatement(); + + try { + stmt.execute("drop table TEST_NULL_INT"); + } catch (final SQLException sqle) { + } + + stmt.execute("create table TEST_NULL_INT (id integer not null, val1 varchar(50), constraint my_pk primary key (id))"); + + // Insert some valid numeric values (for TO_NUMBER call later) + for (int i = 0; i < 11; i++) { + stmt.execute("insert into TEST_NULL_INT (id, val1) VALUES (" + i + ", '" + i + "')"); + } + // Insert invalid numeric value + stmt.execute("insert into TEST_NULL_INT (id, val1) VALUES (100, 'abc')"); + + Map<String, String> attrMap = new HashMap<>(); + String testAttrName = "attr1"; + String testAttrValue = "value1"; + attrMap.put(testAttrName, testAttrValue); + + MockRecordWriter recordWriter = new MockRecordWriter(null, true, -1); + runner.addControllerService("writer", recordWriter); + runner.setProperty(ExecuteSQLRecord.RECORD_WRITER_FACTORY, "writer"); + runner.enableControllerService(recordWriter); + + runner.setIncomingConnection(true); + runner.setProperty(ExecuteSQLRecord.MAX_ROWS_PER_FLOW_FILE, "5"); + runner.enqueue("SELECT ID, CAST(VAL1 AS INTEGER) AS TN FROM TEST_NULL_INT", attrMap); + runner.run(); + + runner.assertAllFlowFilesTransferred(ExecuteSQLRecord.REL_FAILURE, 1); + runner.assertTransferCount(ExecuteSQLRecord.REL_SUCCESS, 0); + } + + @Test public void testMaxRowsPerFlowFile() throws Exception { // remove previous test database, if any final File dbLocation = new File(DB_LOCATION);
