This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push: new bc95799a39 NIFI-13397 Updated PutDatabaseRecord to retry transient ProcessException causes bc95799a39 is described below commit bc95799a397eaf1fda6a0db4927724b85bd2a496 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-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java index e0dbf5b38b..3210c609a4 100644 --- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java +++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java @@ -585,7 +585,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-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java index f2678bb3ae..dd004e7db7 100644 --- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/PutDatabaseRecordTest.java +++ b/nifi-extension-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; @@ -263,6 +264,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()); @@ -2335,6 +2387,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;