Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-8146: Added RecordPath for Statement Type and to specify where to find the data within the incoming Record. Also performed some minor code refactoring to avoid lots of large lambdas, as it results in code that is more readable and more maintainable. Updated some properties to use dependsOn() notation now that this is available. 

NIFI-8146: Updated PutDatabaseRecord to avoid using the functional style framework that had previously been used. Doing so resulted in code that was difficult to understand and maintain. Added additional unit tests and improved MockRecordParser so that it could throw configurable types of exceptions

NIFI-8146: Fixed checkstyle issues

Signed-off-by: Matthew Burgess <mattyb149@apache.org>

This closes #4763
This commit is contained in:
Mark Payne 2021-01-15 14:38:12 -05:00 committed by Matthew Burgess
parent 3e72dfe421
commit f02079c2fa
No known key found for this signature in database
GPG Key ID: 05D3DEB8126DAD24
4 changed files with 587 additions and 376 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-mock-record-utils/src/main/java/org/apache/nifi/serialization/record/MockRecordFailureType.java Normal file
View File

@ -0,0 +1,24 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.nifi.serialization.record;
public enum MockRecordFailureType {
IO_EXCEPTION,
MALFORMED_RECORD_EXCEPTION;
}

12
nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-mock-record-utils/src/main/java/org/apache/nifi/serialization/record/MockRecordParser.java
View File

@ -37,6 +37,7 @@ public class MockRecordParser extends AbstractControllerService implements Recor
private final List<Object[]> records = new ArrayList<>();
private final List<RecordField> fields = new ArrayList<>();
private int failAfterN;
private MockRecordFailureType failureType = MockRecordFailureType.MALFORMED_RECORD_EXCEPTION;
public MockRecordParser() {
this(-1);
@ -47,7 +48,12 @@ public class MockRecordParser extends AbstractControllerService implements Recor
}
public void failAfter(final int failAfterN) {
failAfter(failAfterN, MockRecordFailureType.MALFORMED_RECORD_EXCEPTION);
}
public void failAfter(final int failAfterN, final MockRecordFailureType failureType) {
this.failAfterN = failAfterN;
this.failureType = failureType;
}
public void addSchemaField(final String fieldName, final RecordFieldType type) {
@ -80,7 +86,11 @@ public class MockRecordParser extends AbstractControllerService implements Recor
@Override
public Record nextRecord(final boolean coerceTypes, final boolean dropUnknown) throws IOException, MalformedRecordException {
if (failAfterN >= 0 && recordCount >= failAfterN) {
throw new MalformedRecordException("Intentional Unit Test Exception because " + recordCount + " records have been read");
if (failureType == MockRecordFailureType.MALFORMED_RECORD_EXCEPTION) {
throw new MalformedRecordException("Intentional Unit Test Exception because " + recordCount + " records have been read");
} else {
throw new IOException("Intentional Unit Test Exception because " + recordCount + " records have been read");
}
}
recordCount++;

760
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java
View File

File diff suppressed because it is too large Load Diff

167
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/TestPutDatabaseRecord.groovy
View File

@ -16,11 +16,13 @@
*/
package org.apache.nifi.processors.standard
import org.apache.commons.dbcp2.DelegatingConnection
import org.apache.nifi.processor.exception.ProcessException
import org.apache.nifi.processor.util.pattern.RollbackOnFailure
import org.apache.nifi.reporting.InitializationException
import org.apache.nifi.serialization.SimpleRecordSchema
import org.apache.nifi.serialization.record.MapRecord
import org.apache.nifi.serialization.record.MockRecordFailureType
import org.apache.nifi.serialization.record.MockRecordParser
import org.apache.nifi.serialization.record.RecordField
import org.apache.nifi.serialization.record.RecordFieldType
@ -52,6 +54,7 @@ import static org.junit.Assert.assertNotNull
import static org.junit.Assert.assertNull
import static org.junit.Assert.assertTrue
import static org.junit.Assert.fail
import static org.mockito.ArgumentMatchers.anyMap
import static org.mockito.Mockito.doAnswer
import static org.mockito.Mockito.spy
@ -119,9 +122,9 @@ class TestPutDatabaseRecord {
void testGeneratePreparedStatements() throws Exception {
final List<RecordField> fields = [new RecordField('id', RecordFieldType.INT.dataType),
new RecordField('name', RecordFieldType.STRING.dataType),
new RecordField('code', RecordFieldType.INT.dataType),
new RecordField('non_existing', RecordFieldType.BOOLEAN.dataType)]
new RecordField('name', RecordFieldType.STRING.dataType),
new RecordField('code', RecordFieldType.INT.dataType),
new RecordField('non_existing', RecordFieldType.BOOLEAN.dataType)]
def schema = [
getFields : {fields},
@ -147,8 +150,8 @@ class TestPutDatabaseRecord {
runner.setProperty(PutDatabaseRecord.TRANSLATE_FIELD_NAMES, 'false')
runner.setProperty(PutDatabaseRecord.UNMATCHED_FIELD_BEHAVIOR, PutDatabaseRecord.IGNORE_UNMATCHED_FIELD)
runner.setProperty(PutDatabaseRecord.UNMATCHED_COLUMN_BEHAVIOR, PutDatabaseRecord.IGNORE_UNMATCHED_COLUMN)
runner.setProperty(PutDatabaseRecord.QUOTED_IDENTIFIERS, 'false')
runner.setProperty(PutDatabaseRecord.QUOTED_TABLE_IDENTIFIER, 'false')
runner.setProperty(PutDatabaseRecord.QUOTE_IDENTIFIERS, 'false')
runner.setProperty(PutDatabaseRecord.QUOTE_TABLE_IDENTIFIER, 'false')
def settings = new PutDatabaseRecord.DMLSettings(runner.getProcessContext())
processor.with {
@ -196,8 +199,8 @@ class TestPutDatabaseRecord {
runner.setProperty(PutDatabaseRecord.TRANSLATE_FIELD_NAMES, 'false')
runner.setProperty(PutDatabaseRecord.UNMATCHED_FIELD_BEHAVIOR, PutDatabaseRecord.FAIL_UNMATCHED_FIELD)
runner.setProperty(PutDatabaseRecord.UNMATCHED_COLUMN_BEHAVIOR, PutDatabaseRecord.IGNORE_UNMATCHED_COLUMN)
runner.setProperty(PutDatabaseRecord.QUOTED_IDENTIFIERS, 'false')
runner.setProperty(PutDatabaseRecord.QUOTED_TABLE_IDENTIFIER, 'false')
runner.setProperty(PutDatabaseRecord.QUOTE_IDENTIFIERS, 'false')
runner.setProperty(PutDatabaseRecord.QUOTE_TABLE_IDENTIFIER, 'false')
def settings = new PutDatabaseRecord.DMLSettings(runner.getProcessContext())
processor.with {
@ -292,7 +295,7 @@ class TestPutDatabaseRecord {
parser.addRecord(1, 'rec1', 101)
parser.addRecord(2, 'rec2', 102)
parser.addRecord(3, 'rec3', 1000)
parser.addRecord(3, 'rec3', 1000) // This record violates the constraint on the 'code' column so should result in FlowFile being routed to failure
parser.addRecord(4, 'rec4', 104)
runner.setProperty(PutDatabaseRecord.RECORD_READER_FACTORY, 'parser')
@ -302,9 +305,7 @@ class TestPutDatabaseRecord {
runner.enqueue(new byte[0])
runner.run()
runner.assertTransferCount(PutDatabaseRecord.REL_SUCCESS, 0)
runner.assertTransferCount(PutDatabaseRecord.REL_FAILURE, 0)
runner.assertTransferCount(PutDatabaseRecord.REL_RETRY, 1)
runner.assertAllFlowFilesTransferred(PutDatabaseRecord.REL_FAILURE, 1)
final Connection conn = dbcp.getConnection()
final Statement stmt = conn.createStatement()
final ResultSet rs = stmt.executeQuery('SELECT * FROM PERSONS')
@ -337,12 +338,7 @@ class TestPutDatabaseRecord {
runner.setProperty(RollbackOnFailure.ROLLBACK_ON_FAILURE, 'true')
runner.enqueue(new byte[0])
try {
runner.run()
fail("ProcessException is expected")
} catch (AssertionError e) {
assertTrue(e.getCause() instanceof ProcessException)
}
runner.run()
final Connection conn = dbcp.getConnection()
final Statement stmt = conn.createStatement()
@ -494,6 +490,76 @@ class TestPutDatabaseRecord {
conn.close()
}
@Test
void testInvalidData() throws InitializationException, ProcessException, SQLException, IOException {
recreateTable("PERSONS", createPersons)
final MockRecordParser parser = new MockRecordParser()
runner.addControllerService("parser", parser)
runner.enableControllerService(parser)
parser.addSchemaField("id", RecordFieldType.INT)
parser.addSchemaField("name", RecordFieldType.STRING)
parser.addSchemaField("code", RecordFieldType.INT)
parser.addRecord(1, 'rec1', 101)
parser.addRecord(2, 'rec2', 102)
parser.addRecord(3, 'rec3', 104)
parser.failAfter(1)
runner.setProperty(PutDatabaseRecord.RECORD_READER_FACTORY, 'parser')
runner.setProperty(PutDatabaseRecord.STATEMENT_TYPE, PutDatabaseRecord.INSERT_TYPE)
runner.setProperty(PutDatabaseRecord.TABLE_NAME, 'PERSONS')
runner.enqueue(new byte[0])
runner.run()
runner.assertAllFlowFilesTransferred(PutDatabaseRecord.REL_FAILURE, 1)
final Connection conn = dbcp.getConnection()
final Statement stmt = conn.createStatement()
final ResultSet rs = stmt.executeQuery('SELECT * FROM PERSONS')
// Transaction should be rolled back and table should remain empty.
assertFalse(rs.next())
stmt.close()
conn.close()
}
@Test
void testIOExceptionOnReadData() throws InitializationException, ProcessException, SQLException, IOException {
recreateTable("PERSONS", createPersons)
final MockRecordParser parser = new MockRecordParser()
runner.addControllerService("parser", parser)
runner.enableControllerService(parser)
parser.addSchemaField("id", RecordFieldType.INT)
parser.addSchemaField("name", RecordFieldType.STRING)
parser.addSchemaField("code", RecordFieldType.INT)
parser.addRecord(1, 'rec1', 101)
parser.addRecord(2, 'rec2', 102)
parser.addRecord(3, 'rec3', 104)
parser.failAfter(1, MockRecordFailureType.IO_EXCEPTION)
runner.setProperty(PutDatabaseRecord.RECORD_READER_FACTORY, 'parser')
runner.setProperty(PutDatabaseRecord.STATEMENT_TYPE, PutDatabaseRecord.INSERT_TYPE)
runner.setProperty(PutDatabaseRecord.TABLE_NAME, 'PERSONS')
runner.enqueue(new byte[0])
runner.run()
runner.assertAllFlowFilesTransferred(PutDatabaseRecord.REL_FAILURE, 1)
final Connection conn = dbcp.getConnection()
final Statement stmt = conn.createStatement()
final ResultSet rs = stmt.executeQuery('SELECT * FROM PERSONS')
// Transaction should be rolled back and table should remain empty.
assertFalse(rs.next())
stmt.close()
conn.close()
}
@Test
void testSqlStatementTypeNoValue() throws InitializationException, ProcessException, SQLException, IOException {
recreateTable("PERSONS", createPersons)
@ -539,12 +605,8 @@ class TestPutDatabaseRecord {
def attrs = [:]
attrs[PutDatabaseRecord.STATEMENT_TYPE_ATTRIBUTE] = 'sql'
runner.enqueue(new byte[0], attrs)
try {
runner.run()
fail("ProcessException is expected")
} catch (AssertionError e) {
assertTrue(e.getCause() instanceof ProcessException)
}
runner.run()
runner.assertTransferCount(PutDatabaseRecord.REL_SUCCESS, 0)
runner.assertTransferCount(PutDatabaseRecord.REL_FAILURE, 0)
@ -666,6 +728,7 @@ class TestPutDatabaseRecord {
runner.setProperty(PutDatabaseRecord.STATEMENT_TYPE, PutDatabaseRecord.UPDATE_TYPE)
runner.setProperty(PutDatabaseRecord.TABLE_NAME, 'PERSONS')
parser.addRecord(1, 'rec1', 201)
runner.enqueue(new byte[0])
runner.run()
@ -809,6 +872,58 @@ class TestPutDatabaseRecord {
conn.close()
}
@Test
void testRecordPathOptions() {
recreateTable("PERSONS", 'CREATE TABLE PERSONS (id integer, name varchar(100), code integer)')
final MockRecordParser parser = new MockRecordParser()
runner.addControllerService("parser", parser)
runner.enableControllerService(parser)
final List<RecordField> dataFields = new ArrayList<>();
dataFields.add(new RecordField("id", RecordFieldType.INT.getDataType()))
dataFields.add(new RecordField("name", RecordFieldType.STRING.getDataType()))
dataFields.add(new RecordField("code", RecordFieldType.INT.getDataType()))
final RecordSchema dataSchema = new SimpleRecordSchema(dataFields)
parser.addSchemaField("operation", RecordFieldType.STRING)
parser.addSchemaField(new RecordField("data", RecordFieldType.RECORD.getRecordDataType(dataSchema)))
// CREATE, CREATE, CREATE, DELETE, UPDATE
parser.addRecord("INSERT", new MapRecord(dataSchema, ["id": 1, "name": "John Doe", "code": 55] as Map))
parser.addRecord("INSERT", new MapRecord(dataSchema, ["id": 2, "name": "Jane Doe", "code": 44] as Map))
parser.addRecord("INSERT", new MapRecord(dataSchema, ["id": 3, "name": "Jim Doe", "code": 2] as Map))
parser.addRecord("DELETE", new MapRecord(dataSchema, ["id": 2, "name": "Jane Doe", "code": 44] as Map))
parser.addRecord("UPDATE", new MapRecord(dataSchema, ["id": 1, "name": "John Doe", "code": 201] as Map))
runner.setProperty(PutDatabaseRecord.RECORD_READER_FACTORY, 'parser')
runner.setProperty(PutDatabaseRecord.STATEMENT_TYPE, PutDatabaseRecord.USE_RECORD_PATH)
runner.setProperty(PutDatabaseRecord.DATA_RECORD_PATH, "/data")
runner.setProperty(PutDatabaseRecord.STATEMENT_TYPE_RECORD_PATH, "/operation")
runner.setProperty(PutDatabaseRecord.UPDATE_KEYS, 'id')
runner.setProperty(PutDatabaseRecord.TABLE_NAME, 'PERSONS')
runner.enqueue(new byte[0])
runner.run()
runner.assertAllFlowFilesTransferred(PutDatabaseRecord.REL_SUCCESS, 1)
Connection conn = dbcp.getConnection()
final Statement stmt = conn.createStatement()
final ResultSet rs = stmt.executeQuery('SELECT * FROM PERSONS')
assertTrue(rs.next())
assertEquals(1, rs.getInt(1))
assertEquals('John Doe', rs.getString(2))
assertEquals(201, rs.getInt(3))
assertTrue(rs.next())
assertEquals(3, rs.getInt(1))
assertEquals('Jim Doe', rs.getString(2))
assertEquals(2, rs.getInt(3))
assertFalse(rs.next())
stmt.close()
conn.close()
}
@Test
void testInsertWithMaxBatchSize() throws InitializationException, ProcessException, SQLException, IOException {
recreateTable("PERSONS", createPersons)
@ -947,8 +1062,8 @@ class TestPutDatabaseRecord {
runner.setProperty(PutDatabaseRecord.TRANSLATE_FIELD_NAMES, 'false')
runner.setProperty(PutDatabaseRecord.UNMATCHED_FIELD_BEHAVIOR, PutDatabaseRecord.IGNORE_UNMATCHED_FIELD)
runner.setProperty(PutDatabaseRecord.UNMATCHED_COLUMN_BEHAVIOR, PutDatabaseRecord.IGNORE_UNMATCHED_COLUMN)
runner.setProperty(PutDatabaseRecord.QUOTED_IDENTIFIERS, 'true')
runner.setProperty(PutDatabaseRecord.QUOTED_TABLE_IDENTIFIER, 'true')
runner.setProperty(PutDatabaseRecord.QUOTE_IDENTIFIERS, 'true')
runner.setProperty(PutDatabaseRecord.QUOTE_TABLE_IDENTIFIER, 'true')
def settings = new PutDatabaseRecord.DMLSettings(runner.getProcessContext())
processor.with {