[BUG] update and upsert bulk actions do not include changes from document_root_key, exclude_keys, etc

[graytaylor0]

Describe the bug
The update and upsert actions always pass the full Event to the bulk actions, instead of only passing the part of the Event that is filtered from document_root_key, exclude_keys, include_keys, etc.

For update we are passing the original jsonNode here (

data-prepper/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

Line 307 in 979a004

.document(jsonNode)

), while for index we pass the document without the root key / exclude_keys here (

data-prepper/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

Line 334 in 979a004

.document(document)

).

To Reproduce
Steps to reproduce the behavior:

Send two items with the following structure and sink config

{"operation": "index", "item": { "doc_id": 110, "name": "Original Name" }}
{"operation": "update", "item": { "doc_id": 110, "name": "Updated Name" }}

        document_id_field: "item/doc_id"
        index: "actions-test-index"
        document_root_key: "item"
        actions:
          - type: "update"
            when: '/operation == "update"'
          - type: "index"

and the final document in OpenSearch will look like

{
        "_index": "actions-test-index",
        "_id": "110",
        "_score": 1,
        "_source": {
          "doc_id": 110,
          "name": "Original Name",
          "item": {
            "name": "Updated Name",
            "doc_id": 110
          },
          "operation": "update"
        }
}

Expected behavior

The updated document should look like

{
        "_index": "actions-test-index",
        "_id": "110",
        "_score": 1,
        "_source": {
          "doc_id": 110,
          "name": "Updated Name"
        }
}

[dlvenable]

The update API in opensearch-java requires we pass in a JsonNode. So that part is what we need. Perhaps the problem is from where we get the JsonNode. Could we get this from the SerializedDocument object instead?

[dlvenable]

Also, I think this would be a good opportunity to split this logic into another class. There is too much going on in there. If we had a class responsible for this mapping we could have some really solid unit tests to ensure that the output is what we expect for each case.