This is an automated email from the ASF dual-hosted git repository. mattyb149 pushed a commit to branch support/nifi-1.x in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push: new 4d6e55ba37 NIFI-13397 Updated PutDatabaseRecord to retry transient ProcessException causes 4d6e55ba37 is described below commit 4d6e55ba37ddb65e79f610d2b081a5e07241bd40 Author: Jim Steinebrey <jrsteineb...@gmail.com> AuthorDate: Thu Jun 13 12:27:17 2024 -0400 NIFI-13397 Updated PutDatabaseRecord to retry transient ProcessException causes This closes #8964 Signed-off-by: David Handermann <exceptionfact...@apache.org> --- .../processors/standard/PutDatabaseRecord.java | 2 +- .../processors/standard/PutDatabaseRecordTest.java | 70 ++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java index c279569916..3be6eb5d33 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java @@ -590,7 +590,7 @@ public class PutDatabaseRecord extends AbstractProcessor { // When an Exception is thrown, we want to route to 'retry' if we expect that attempting the same request again // might work. Otherwise, route to failure. SQLTransientException is a specific type that indicates that a retry may work. final Relationship relationship; - final Throwable toAnalyze = (e instanceof BatchUpdateException) ? e.getCause() : e; + final Throwable toAnalyze = (e instanceof BatchUpdateException || e instanceof ProcessException) ? e.getCause() : e; if (toAnalyze instanceof SQLTransientException) { relationship = REL_RETRY; flowFile = session.penalize(flowFile); diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java index 733b109e80..e56a3c5b77 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java @@ -58,6 +58,7 @@ import java.sql.ResultSet; import java.sql.SQLDataException; import java.sql.SQLException; import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLTransientException; import java.sql.Statement; import java.time.LocalDate; import java.time.ZoneOffset; @@ -269,6 +270,57 @@ public class PutDatabaseRecordTest { } @Test + public void testProcessExceptionRouteRetry() throws InitializationException, SQLException { + setRunner(TestCaseEnum.DEFAULT_1.getTestCase()); + + // This exception should route to REL_RETRY because its cause is SQLTransientException + dbcp = new DBCPServiceThrowConnectionException(new SQLTransientException("connection failed")); + final Map<String, String> dbcpProperties = new HashMap<>(); + runner = TestRunners.newTestRunner(processor); + runner.addControllerService(DBCP_SERVICE_ID, dbcp, dbcpProperties); + runner.enableControllerService(dbcp); + runner.setProperty(PutDatabaseRecord.DBCP_SERVICE, DBCP_SERVICE_ID); + + final MockRecordParser parser = new MockRecordParser(); + runner.addControllerService("parser", parser); + runner.enableControllerService(parser); + + 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_RETRY); + } + + @Test + public void testProcessExceptionRouteFailure() throws InitializationException, SQLException { + setRunner(TestCaseEnum.DEFAULT_1.getTestCase()); + + // This exception should route to REL_FAILURE because its cause is NOT SQLTransientException + dbcp = new DBCPServiceThrowConnectionException(new NullPointerException("connection is null")); + final Map<String, String> dbcpProperties = new HashMap<>(); + runner = TestRunners.newTestRunner(processor); + runner.addControllerService(DBCP_SERVICE_ID, dbcp, dbcpProperties); + runner.enableControllerService(dbcp); + runner.setProperty(PutDatabaseRecord.DBCP_SERVICE, DBCP_SERVICE_ID); + + final MockRecordParser parser = new MockRecordParser(); + runner.addControllerService("parser", parser); + runner.enableControllerService(parser); + + 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); + } + public void testInsertNonRequiredColumnsUnmatchedField() throws InitializationException, ProcessException { setRunner(TestCaseEnum.DEFAULT_2.getTestCase()); @@ -2349,6 +2401,24 @@ public class PutDatabaseRecordTest { } } + static class DBCPServiceThrowConnectionException extends AbstractControllerService implements DBCPService { + private final Exception rootCause; + + public DBCPServiceThrowConnectionException(final Exception rootCause) { + this.rootCause = rootCause; + } + + @Override + public String getIdentifier() { + return DBCP_SERVICE_ID; + } + + @Override + public Connection getConnection() throws ProcessException { + throw new ProcessException(rootCause); + } + } + static class DBCPServiceAutoCommitTest extends AbstractControllerService implements DBCPService { private final String databaseLocation;