[8.3] Don't ignore pipeline for upserts in bulk api (#87719) (#88054)

* Don't ignore pipeline for upserts in bulk api (#87719)

Don't ignore pipelines defined in bulk item for updates with upsert in bulk api.

Closes #87131

* update skip version section

docs/changelog/87719.yaml

@@ -0,0 +1,6 @@
+pr: 87719
+summary: Don't ignore pipeline for upserts in bulk api
+area: Ingest
+type: bug
+issues:
+ - 87131

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

@@ -144,3 +144,40 @@ teardown:
 
   - is_false: _source.field1
   - match: {_source.field2: value2}
+
+---
+"Update with pipeline":
+  - skip:
+      version: " - 7.17.4"
+      reason: "fixed in 7.17.5"
+
+  - do:
+      ingest.put_pipeline:
+        id: "lowercase-pipeline"
+        body: >
+          {
+            "processors": [
+              {
+                "lowercase" : {
+                  "field" : "my_field"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"update":{"_id":"1","_index":"test_index","pipeline":"lowercase-pipeline"}}'
+          - '{"upsert":{"my_field":"UPPER"},"script":{"source":"ctx._source.my_field = ctx._source.my_field.toLowercase()"}}'
+
+  - match: { errors: false }
+  - match: { items.0.update.result: created }
+
+  - do:
+      get:
+        index: test_index
+        id: 1
+  - match: { _source:  { my_field: "upper" } }

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

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