NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord #6903

ChrisSamo632 · 2023-01-29T19:34:05Z

Summary

NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord

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 8
- 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~~

...cessors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java

MikeThomsen · 2023-02-22T14:28:33Z

...cessors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java

+                    try (final OutputStream errorsOutputStream = session.write(errorResponsesFF)) {
+                        errorMapper.writeValue(errorsOutputStream, errorResponses);
+
+                        errorResponsesFF = session.putAttribute(errorResponsesFF, "elasticsearch.put.error.count", String.valueOf(errorResponses.size()));


Have you tested this live? I've had problems with this in the past where the ProcessSession didn't want to update the attributes while the OutputStream was open.

Seemed to work OK when I tried it, but should be able to move this outside of the try-with-resources if you've seen it cause problems in the past

I'll test it out locally to verify. I think I had that problem on 1.13, so it may be fixed by now.

To be fair, looking at the code it would make more sense for the transfer and attributes calls to be outside of the try-with-resources, so I've made the change and will push it once all the tests have passed locally 👍

...src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearchTest.groovy

ChrisSamo632 · 2023-03-20T20:46:46Z

@mattyb149 I believe I addressed your comments

@MikeThomsen do the updates help with clarity around the transfer of FlowFiles/assignment of attributes?

mattyb149 · 2023-03-21T03:54:20Z

Sorry I lost track of this, @MikeThomsen I'm good if you are

davis-anthony · 2023-04-18T22:38:22Z

I've been struggling with the various PutElasticsearch processors and their different handling of responses/errors for a while now, and nothing really does what I would expect it to do.

The only one that kind of works is PutElasticsearchHttp because at least that one places the "reason" attribute on the flowfile, but it doesn't handle all types of errors and it makes a bunch of assumptions as to what is retryable.

What I really need is for the original documents that fail on a bulk index/create to be routed to failure or error and have the error message and status code added as flowfile attributes. This would enable custom routing logic based on status code and/or error message. I can't tell if that's what this PR is attempting to do. Is that what this PR does or should I post a new feature request for this?

If it's just sending the error response documents as separate flowfiles I can see how some developers might find that useful, but it doesn't really help with rerouting the original document and providing error information.

…File to PutElasticsearchJson and PutElasticsearchRecord

…son/Record processors

ChrisSamo632 · 2023-04-19T11:14:32Z

@davis-anthony this PR if/when Approved (@MikeThomsen / @mattyb149 ) was not going to introduce such behaviour, but it does make sense for PutElasticsearchJson so I've added a new elasticsearch.bulk.error attribute for files being sent to the errors relationship - this will contain the _bulk API's response for the document if Elasticsearch has marked it as errored (and if it's not_found if you set the Treat "Not Found" as Success to false)

It's not as simple for PutElasticsearchRecord because the errors output FlowFile may contain multiple records (each being a document sent to Elasticsearch). So we'd either be serialising all errors into a single attribute that could be huge (and probably break attribute value limits) or adding an attribute for every single record, which would cause memory issues in NiFi. An alternative would be to produce a single errors FlowFile for every errored Record from the input FlowFile, which again would cause performance problems in NiFi if you're trying to process large amounts of Records (which is the big benefit of Record-based processors, e.g. millions of records within a single file).

A flow I've used before for handling things like errors from a record processor such as PutElasticsearchRecord is to send the errors to PutDistributedMapCache keyed on the document _id (which is in the error response from Elasticsearch) and then using the FetchDistributedMapCache or LookupAttribute to enrich each of the records in the PutElasticsearchRecord output in the cases where there's an _id match - it's a bit fidly and could require splitting FlowFiles by Record (which again brings us back to the performance hit mentioned above).

Notes:

PutElasticsearchHttp (and PutElasticsearchHttpRecord) are deprecated is recent 1.x versions of NiFi and will be removed in NiFi 2.x
the elasticsearch.put.error attribute for both PutElasticsearchJson and PutElasticsearchRecord are used for general Elasticsearch connection error reporting, e.g. if the Elasticsearch instance/cluster cis not found or authentication/authorisation fails, etc., and FlowFile processing, e.g. if the content of the FlowFile sent to PutElastichsearchJson can't be parsed as a JSON object

…_bulk response for error documents in PutElasticsearchJson

davis-anthony · 2023-04-21T13:49:20Z

@ChrisSamo632 Thanks for the explanation and for helping me understand the issues with memory. Yes that makes a lot of sense and after reading the code I see that the processor makes every effort to just pass the documents to the various relationships. So with that in mind I propose that we make these processors behave like the RouteOnAttribute processor where the developer can define additional relationships based on expected error types. So that the processor can simply compare the error type for each item in the collection and route to the various relationships.

For example, I might define the following:

conflict -> version_conflict_engine_exception
schema -> mapper_parsing_exception
blocked -> cluster_block_exception

Then the following relationships could be routed in the flow: success, error, retry, failure, conflict, schema, blocked.

The result would be that errors that match the 3 types are routed to 3 custom routes. The remaining errors and other error routes would operate as they do now.

The only flaw I can think of here is that there is no definitive list of error types according to the following post, and so it takes quite a bit of sleuthing to find the possible errors and they are likely subject to change since there is no list.
https://discuss.elastic.co/t/bulk-api-item-error-types/242639

ChrisSamo632 · 2023-04-21T14:16:19Z

@davis-anthony that would definitely warrant a new Jira ticket to make such a change I think.

Some sort of "half-way house" approach may be more appropriate, e.g. like PartitionRecord where errors Records are grouped together into a FlowFile coming from PutElasticsearchRecord based upon the type of error indicated by the Elasticsearch response, with the type added to the FlowFile as the elasticsearch.bulk.error attribute, something like that might be helpful and hopefully avoid splitting FlowFiles up too much. This would also be an optional thing to enable in the processor.

Feel free to raise a ticket for further discussion.

ChrisSamo632 · 2023-04-24T22:03:06Z

NIFI-11480 for the above ☝️ discussion, FYI @davis-anthony

MikeThomsen · 2023-04-24T23:47:08Z

...cessors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java

    static final Relationship REL_SUCCESS = new Relationship.Builder()
            .name("success")
            .description("All flowfiles that succeed in being transferred into Elasticsearch go here. " +
                    "Documents received by the Elasticsearch _bulk API may still result in errors on the Elasticsearch side. " +
                    "The Elasticsearch response will need to be examined to determine whether any Document(s)/Record(s) resulted in errors.")
            .build();

+    static final Relationship REL_ERROR_RESPONSES = new Relationship.Builder()
+            .name("error_responses")


This should be autoterminated because it changes the behavior.

…File to PutElasticsearchJson and PutElasticsearchRecord NIFI-11111 clarify error_responses relationships in PutElasticsearchJson/Record processors NIFI-11111 Refactor exception handling for error response flowfile transfer NIFI-11111 Add elasticsearch.bulk.error attributes containing the Elasticsearch _bulk response for error documents in PutElasticsearchJson This closes #6903 Signed-off-by: Mike Thomsen <mthomsen@apache.org>

ChrisSamo632 force-pushed the NIFI-11111 branch 2 times, most recently from 653d4b3 to 7976f05 Compare February 5, 2023 17:02

mattyb149 reviewed Feb 7, 2023

View reviewed changes

...cessors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java Outdated Show resolved Hide resolved

...cessors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java Outdated Show resolved Hide resolved

ChrisSamo632 force-pushed the NIFI-11111 branch 2 times, most recently from 862428e to cc9622f Compare February 10, 2023 21:13

MikeThomsen requested changes Feb 22, 2023

View reviewed changes

ChrisSamo632 force-pushed the NIFI-11111 branch from cc9622f to a73f82d Compare February 22, 2023 19:33

ChrisSamo632 added 3 commits April 19, 2023 09:11

NIFI-11111 add option to output Elasticsearch error responses as Flow…

bf33e42

…File to PutElasticsearchJson and PutElasticsearchRecord

NIFI-11111 clarify error_responses relationships in PutElasticsearchJ…

a011edf

…son/Record processors

Refactor exception handling for error response flowfile transfer

9dc7d40

ChrisSamo632 force-pushed the NIFI-11111 branch 2 times, most recently from ac9454c to 8aafcfb Compare April 19, 2023 17:36

Add elasticsearch.bulk.error attributes containing the Elasticsearch …

037d50d

…_bulk response for error documents in PutElasticsearchJson

ChrisSamo632 force-pushed the NIFI-11111 branch from 8aafcfb to 037d50d Compare April 19, 2023 17:50

MikeThomsen approved these changes Apr 25, 2023

View reviewed changes

asfgit closed this in 05418d9 Apr 25, 2023

ChrisSamo632 deleted the NIFI-11111 branch May 2, 2023 16:01

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord #6903

NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord #6903

ChrisSamo632 commented Jan 29, 2023 •

edited

Loading

MikeThomsen Feb 22, 2023

ChrisSamo632 Feb 22, 2023

MikeThomsen Feb 22, 2023

ChrisSamo632 Feb 22, 2023

ChrisSamo632 commented Mar 20, 2023

mattyb149 commented Mar 21, 2023

davis-anthony commented Apr 18, 2023

ChrisSamo632 commented Apr 19, 2023

davis-anthony commented Apr 21, 2023

ChrisSamo632 commented Apr 21, 2023

ChrisSamo632 commented Apr 24, 2023 •

edited

Loading

MikeThomsen Apr 24, 2023

NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord #6903

NIFI-11111 add option to output Elasticsearch error responses as FlowFile to PutElasticsearchJson and PutElasticsearchRecord #6903

Conversation

ChrisSamo632 commented Jan 29, 2023 • edited Loading

Summary

Tracking

Issue Tracking

Pull Request Tracking

Pull Request Formatting

Verification

Build

Licensing

Documentation

MikeThomsen Feb 22, 2023

Choose a reason for hiding this comment

ChrisSamo632 Feb 22, 2023

Choose a reason for hiding this comment

MikeThomsen Feb 22, 2023

Choose a reason for hiding this comment

ChrisSamo632 Feb 22, 2023

Choose a reason for hiding this comment

ChrisSamo632 commented Mar 20, 2023

mattyb149 commented Mar 21, 2023

davis-anthony commented Apr 18, 2023

ChrisSamo632 commented Apr 19, 2023

davis-anthony commented Apr 21, 2023

ChrisSamo632 commented Apr 21, 2023

ChrisSamo632 commented Apr 24, 2023 • edited Loading

MikeThomsen Apr 24, 2023

Choose a reason for hiding this comment

ChrisSamo632 commented Jan 29, 2023 •

edited

Loading

ChrisSamo632 commented Apr 24, 2023 •

edited

Loading