Fix bulk API ignores ingest pipeline for upsert

CHANGELOG.md

@@ -119,6 +119,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Removed
 
 ### Fixed
+- Fix bulk API ignores ingest pipeline for upsert ([#12883](https://github.com/opensearch-project/OpenSearch/pull/12883))
 - Fix issue with feature flags where default value may not be honored ([#12849](https://github.com/opensearch-project/OpenSearch/pull/12849))
 
 ### Security

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml

@@ -144,3 +144,26 @@ teardown:
 
   - is_false: _source.field1
   - match: {_source.field2: value2}
+
+# related issue: https://github.com/opensearch-project/OpenSearch/issues/12854
+---
+"Test bulk honors pipeline in update action with upsert":
+  - skip:
+      version: " - 2.99.99"
+      reason: "fixed in 3.0.0"
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"update": {"_index": "test_index", "_id": "test_id3", "pipeline": "pipeline1"}}'
+          - '{"upsert": {"f1": "v2", "f2": 47}, "doc": {"x": 1}}'
+
+  - match: { errors: false }
+  - match: { items.0.update.result: created }
+
+  - do:
+      get:
+        index: test_index
+        id: test_id3
+  - match: { _source: {"f1": "v2", "f2": 47, "field1": "value1"}}

server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java

@@ -371,7 +371,7 @@ public void parse(
                         }
                         IndexRequest upsertRequest = updateRequest.upsertRequest();
                         if (upsertRequest != null) {
-                            upsertRequest.setPipeline(defaultPipeline);
+                            upsertRequest.setPipeline(pipeline);
                         }
 
                         updateRequestConsumer.accept(updateRequest);

server/src/test/java/org/opensearch/action/bulk/BulkRequestParserTests.java

@@ -112,6 +112,13 @@ public void testUpdateRequest() throws IOException {
         parser.parse(request, "foo", null, null, null, true, false, MediaTypeRegistry.JSON, req -> fail(), updateRequest -> {
             assertFalse(updateRequest.isRequireAlias());
         }, req -> fail());
+
+        request = new BytesArray(
+            "{ \"update\":{ \"_id\": \"bar\", \"require_alias\": false, \"pipeline\": \"testPipeline\" } }\n{\"upsert\": {\"x\": 1}}\n"
+        );
+        parser.parse(request, "foo", null, null, null, true, false, MediaTypeRegistry.JSON, req -> fail(), updateRequest -> {
+            assertEquals(updateRequest.upsertRequest().getPipeline(), "testPipeline");
+        }, req -> fail());
     }
 
     public void testBarfOnLackOfTrailingNewline() {