Repository: nifi Updated Branches: refs/heads/master 8aa445008 -> 50f22162b
NIFI-4286: Fix NPE in GenerateTableFetch when Partition Size = 0 This closes #2078. Signed-off-by: Bryan Bende <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/50f22162 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/50f22162 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/50f22162 Branch: refs/heads/master Commit: 50f22162b06df99094fb8ac99e6520a2ad1d9d92 Parents: 8aa4450 Author: Matt Burgess <[email protected]> Authored: Fri Aug 11 23:38:17 2017 -0400 Committer: Bryan Bende <[email protected]> Committed: Mon Aug 14 15:28:07 2017 -0400 ---------------------------------------------------------------------- .../processors/standard/GenerateTableFetch.java | 6 ++-- .../standard/TestGenerateTableFetch.java | 30 ++++++++++++++++++++ 2 files changed, 33 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/50f22162/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java index 1fcb33e..f5407da 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java @@ -318,12 +318,12 @@ public class GenerateTableFetch extends AbstractDatabaseFetchProcessor { throw new SQLException("No rows returned from metadata query: " + selectQuery); } - final long numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1); + final long numberOfFetches = (partitionSize == 0) ? 1 : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1); // Generate SQL statements to read "pages" of data for (long i = 0; i < numberOfFetches; i++) { - long limit = partitionSize == 0 ? null : partitionSize; - long offset = partitionSize == 0 ? null : i * partitionSize; + Long limit = partitionSize == 0 ? null : (long) partitionSize; + Long offset = partitionSize == 0 ? null : i * partitionSize; final String maxColumnNames = StringUtils.join(maxValueColumnNameList, ", "); final String query = dbAdapter.getSelectStatement(tableName, columnNames, whereClause, maxColumnNames, limit, offset); FlowFile sqlFlowFile = (fileToProcess == null) ? session.create() : session.create(fileToProcess); http://git-wip-us.apache.org/repos/asf/nifi/blob/50f22162/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestGenerateTableFetch.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestGenerateTableFetch.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestGenerateTableFetch.java index 67ab72f..76bc1f0 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestGenerateTableFetch.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestGenerateTableFetch.java @@ -219,6 +219,36 @@ public class TestGenerateTableFetch { } @Test + public void testOnePartition() throws ClassNotFoundException, SQLException, InitializationException, IOException { + + // load test data to database + final Connection con = ((DBCPService) runner.getControllerService("dbcp")).getConnection(); + Statement stmt = con.createStatement(); + + try { + stmt.execute("drop table TEST_QUERY_DB_TABLE"); + } catch (final SQLException sqle) { + // Ignore this error, probably a "table does not exist" since Derby doesn't yet support DROP IF EXISTS [DERBY-4842] + } + + stmt.execute("create table TEST_QUERY_DB_TABLE (id integer not null, bucket integer not null)"); + stmt.execute("insert into TEST_QUERY_DB_TABLE (id, bucket) VALUES (0, 0)"); + stmt.execute("insert into TEST_QUERY_DB_TABLE (id, bucket) VALUES (1, 0)"); + stmt.execute("insert into TEST_QUERY_DB_TABLE (id, bucket) VALUES (2, 0)"); + + runner.setProperty(GenerateTableFetch.TABLE_NAME, "TEST_QUERY_DB_TABLE"); + runner.setIncomingConnection(false); + runner.setProperty(GenerateTableFetch.MAX_VALUE_COLUMN_NAMES, "ID"); + // Set partition size to 0 so we can see that the flow file gets all rows + runner.setProperty(GenerateTableFetch.PARTITION_SIZE, "0"); + + runner.run(); + runner.assertAllFlowFilesTransferred(GenerateTableFetch.REL_SUCCESS, 1); + runner.getFlowFilesForRelationship(GenerateTableFetch.REL_SUCCESS).get(0).assertContentEquals("SELECT * FROM TEST_QUERY_DB_TABLE ORDER BY ID"); + runner.clearTransferState(); + } + + @Test public void testMultiplePartitions() throws ClassNotFoundException, SQLException, InitializationException, IOException { // load test data to database
