diff --git a/CHANGELOG.md b/CHANGELOG.md index 64a52f2c9bf39..69744002b2c9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,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 diff --git a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml b/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml index 2dfa17174b139..d95b1239b1cf2 100644 --- a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml +++ b/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"}} diff --git a/server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java b/server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java index 3fadfe5f2cd6a..f705a218fb8e2 100644 --- a/server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java +++ b/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); diff --git a/server/src/test/java/org/opensearch/action/bulk/BulkRequestParserTests.java b/server/src/test/java/org/opensearch/action/bulk/BulkRequestParserTests.java index 4f07c098b0869..9625475607911 100644 --- a/server/src/test/java/org/opensearch/action/bulk/BulkRequestParserTests.java +++ b/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() {