NIFI-4651 Add error message and code to PutSQL FlowFile attributes by krisztina-zsihovszki · Pull Request #7007 · apache/nifi

krisztina-zsihovszki · 2023-03-03T08:52:21Z

Summary

NIFI-4651

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

Apache NiFi Jira issue created

Pull Request Tracking

Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000

Pull Request Formatting

Pull Request based on current revision of the main branch
Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

Build completed using mvn clean install -P contrib-check
- JDK 11
- JDK 17

Licensing

New dependencies are compatible with the Apache License 2.0 according to the License Policy
New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

Documentation formatting appears as expected in rendered files

krisztina-zsihovszki · 2023-03-03T13:16:02Z

recheck

Lehel44 · 2023-03-09T17:58:31Z

Thanks for the contribution @krisztina-zsihovszki. Reviewing.

Lehel44 · 2023-03-09T23:36:56Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

+            switch (errorTypesResult.destination()) {
                case Failure:
-                    getLogger().error("Failed to update database for {} due to {}; routing to failure", new Object[] {i, e}, e);
+                    getLogger().error("Failed to update database for {} due to {}; routing to failure", new Object[] {flowFile, exception}, exception);


The logger can handle arbitrary number of arguments.

Suggested change

getLogger().error("Failed to update database for {} due to {}; routing to failure", new Object[] {flowFile, exception}, exception);

getLogger().error("Failed to update database for {}; routing to failure", flowFile, exception);

Lehel44 · 2023-03-09T23:39:02Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

                    getLogger().error("Failed to update database for {} due to {}; it is possible that retrying the operation will succeed, so routing to retry",
-                            new Object[] {i, e}, e);
+                            new Object[] {flowFile, exception}, exception);


Lehel44 · 2023-03-09T23:39:11Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

                    break;
                case Self:
-                    getLogger().error("Failed to update database for {} due to {};", new Object[] {i, e}, e);
+                    getLogger().error("Failed to update database for {} due to {};",  new Object[] {flowFile, exception}, exception);


Lehel44 · 2023-03-10T00:23:03Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

+    private ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError(final ProcessContext context, final ProcessSession session, final RoutingResult result) {
+        ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError
+                = ExceptionHandler.createOnGroupError(context, session, result, REL_FAILURE, REL_RETRY);
+        onGroupError = onGroupError.andThen((ctx, flowFileGroup, errorTypesResult, exception) -> {
+
+            switch (errorTypesResult.destination()) {
+                case Failure:
+                    List<FlowFile> flowFilesToFailure = getFlowFilesOnRelationShip(result, REL_FAILURE);
+                    Optional.ofNullable(flowFilesToFailure).map(flowFiles ->
+                        result.getRoutedFlowFiles().put(REL_FAILURE, addErrorAttributesToFlowFilesInGroup(session, flowFiles, flowFileGroup.getFlowFiles(), exception)));
+                   break;
+                case Retry:
+                    List<FlowFile> flowFilesToRetry = getFlowFilesOnRelationShip(result, REL_RETRY);
+                    Optional.ofNullable(flowFilesToRetry).map(flowFiles ->
+                            result.getRoutedFlowFiles().put(REL_RETRY, addErrorAttributesToFlowFilesInGroup(session, flowFiles, flowFileGroup.getFlowFiles(), exception)));
+                    break;
+            }
+        });
+        return onGroupError;
+    }


The RoutingResult::getRoutedFlowFiles already returns an empty map, so the getFlowFilesOnRelationShip does not need to use Optional and should return an empty list in case the flowfiles for the specified relationship are missing. This means that flowFilesToFailure variable in onGroupError cannot be null and also doesn't need to be wrapped in Optionals. I'd also recommend simplifying the 2-branched switch to an if statement for visibility and extracting the condition on getting the Relationship based on the ErrorTypesResult.Destination since both branches depend on that value i.e. -> the if statement is not needed.

Suggested change

private ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError(final ProcessContext context, final ProcessSession session, final RoutingResult result) {

ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError

= ExceptionHandler.createOnGroupError(context, session, result, REL_FAILURE, REL_RETRY);

onGroupError = onGroupError.andThen((ctx, flowFileGroup, errorTypesResult, exception) -> {

switch (errorTypesResult.destination()) {

case Failure:

List<FlowFile> flowFilesToFailure = getFlowFilesOnRelationShip(result, REL_FAILURE);

Optional.ofNullable(flowFilesToFailure).map(flowFiles ->

result.getRoutedFlowFiles().put(REL_FAILURE, addErrorAttributesToFlowFilesInGroup(session, flowFiles, flowFileGroup.getFlowFiles(), exception)));

break;

case Retry:

List<FlowFile> flowFilesToRetry = getFlowFilesOnRelationShip(result, REL_RETRY);

Optional.ofNullable(flowFilesToRetry).map(flowFiles ->

result.getRoutedFlowFiles().put(REL_RETRY, addErrorAttributesToFlowFilesInGroup(session, flowFiles, flowFileGroup.getFlowFiles(), exception)));

break;

}

});

return onGroupError;

}

private ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError(final ProcessContext context, final ProcessSession session, final RoutingResult result) {

ExceptionHandler.OnError<RollbackOnFailure, FlowFileGroup> onGroupError =

ExceptionHandler.createOnGroupError(context, session, result, REL_FAILURE, REL_RETRY);

onGroupError = onGroupError.andThen((ctx, flowFileGroup, errorTypesResult, exception) -> {

Relationship relationship = errorTypesResult.destination() == ErrorTypes.Destination.Failure ? REL_FAILURE : REL_RETRY;

List<FlowFile> flowFilesToRelationship = getFlowFilesOnRelationship(result, relationship);

result.getRoutedFlowFiles().put(relationship, addErrorAttributesToFlowFilesInGroup(session, flowFilesToRelationship, flowFileGroup.getFlowFiles(), exception));

});

return onGroupError;

}

Lehel44 · 2023-03-10T00:24:21Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

+    private List<FlowFile> getFlowFilesOnRelationShip(RoutingResult result, final Relationship relationship) {
+        return Optional.of(result.getRoutedFlowFiles())
+                .orElse(emptyMap())
+                .get(relationship);
+    }


Since the getRoutedFlowFiles already returns an empty map, the optional is not necessary here.

Suggested change

private List<FlowFile> getFlowFilesOnRelationShip(RoutingResult result, final Relationship relationship) {

return Optional.of(result.getRoutedFlowFiles())

.orElse(emptyMap())

.get(relationship);

}

private List<FlowFile> getFlowFilesOnRelationship(RoutingResult result, final Relationship relationship) {

return Optional.ofNullable(result.getRoutedFlowFiles().get(relationship))

.orElse(emptyList());

}

Lehel44 · 2023-03-10T00:34:37Z

...undle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java

+            attributes.put(ERROR_SQL_STATE_ATTR, valueOf(((SQLException) exception).getSQLState()));
+        }
+
+        return session.putAllAttributes(flowFile, attributes);


Is it guaranteed that the flowfile doesn't have original attributes before being updated with the error ones or does the method only update and not replace the flowfile attributes?

The method adds extra attributes to the FlowFile, it does not remove any existing attribute. (Added a check in unit tests to verify this.)

Lehel44

Thanks for making the changes @krisztina-zsihovszki!
LGTM+1

NIFI-4651 Review comments Signed-off-by: Matthew Burgess <mattyb149@apache.org> This closes #7007

mattyb149 · 2023-03-21T15:23:06Z

+1 LGTM, Merging to main and support/nifi-1.x

NIFI-4651 Add error message and code to PutSQL FlowFile attributes

f277264

krisztina-zsihovszki force-pushed the NIFI-4651 branch from fbbe9e8 to f277264 Compare March 7, 2023 09:02

Lehel44 requested changes Mar 10, 2023

View reviewed changes

NIFI-4651 Review comments

f743708

Lehel44 approved these changes Mar 16, 2023

View reviewed changes

mattyb149 pushed a commit that referenced this pull request Mar 21, 2023

NIFI-4651 Add error message and code to PutSQL FlowFile attributes

5267a53

NIFI-4651 Review comments Signed-off-by: Matthew Burgess <mattyb149@apache.org> This closes #7007

mattyb149 closed this Mar 21, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

NIFI-4651 Add error message and code to PutSQL FlowFile attributes#7007

NIFI-4651 Add error message and code to PutSQL FlowFile attributes#7007
krisztina-zsihovszki wants to merge 2 commits intoapache:mainfrom
krisztina-zsihovszki:NIFI-4651

krisztina-zsihovszki commented Mar 3, 2023

Uh oh!

krisztina-zsihovszki commented Mar 3, 2023

Uh oh!

Lehel44 commented Mar 9, 2023

Uh oh!

Lehel44 Mar 9, 2023

Uh oh!

Lehel44 Mar 9, 2023

Uh oh!

Lehel44 Mar 9, 2023

Uh oh!

Lehel44 Mar 10, 2023 •

edited

Loading

Uh oh!

Lehel44 Mar 10, 2023

Uh oh!

Lehel44 Mar 10, 2023

Uh oh!

krisztina-zsihovszki Mar 10, 2023

Uh oh!

Lehel44 left a comment

Uh oh!

mattyb149 commented Mar 21, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	getLogger().error("Failed to update database for {} due to {}; routing to failure", new Object[] {flowFile, exception}, exception);
	getLogger().error("Failed to update database for {}; routing to failure", flowFile, exception);

Conversation

krisztina-zsihovszki commented Mar 3, 2023

Summary

Tracking

Issue Tracking

Pull Request Tracking

Pull Request Formatting

Verification

Build

Licensing

Documentation

Uh oh!

krisztina-zsihovszki commented Mar 3, 2023

Uh oh!

Lehel44 commented Mar 9, 2023

Uh oh!

Lehel44 Mar 9, 2023

Choose a reason for hiding this comment

Uh oh!

Lehel44 Mar 9, 2023

Choose a reason for hiding this comment

Uh oh!

Lehel44 Mar 9, 2023

Choose a reason for hiding this comment

Uh oh!

Lehel44 Mar 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Lehel44 Mar 10, 2023

Choose a reason for hiding this comment

Uh oh!

Lehel44 Mar 10, 2023

Choose a reason for hiding this comment

Uh oh!

krisztina-zsihovszki Mar 10, 2023

Choose a reason for hiding this comment

Uh oh!

Lehel44 left a comment

Choose a reason for hiding this comment

Uh oh!

mattyb149 commented Mar 21, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Lehel44 Mar 10, 2023 •

edited

Loading