Fix class_cast_exception when passing int to _version and other metad…

…ata fields in ingest simulate API (opensearch-project#10101) * Fix class_cast_exception when passing int to _version and other metadata fields in ingest simulate API Signed-off-by: Gao Binlong <[email protected]> * modify change log Signed-off-by: Gao Binlong <[email protected]> * Add more tests Signed-off-by: Gao Binlong <[email protected]> --------- Signed-off-by: Gao Binlong <[email protected]> Signed-off-by: Daniel (dB.) Doubrovkine <[email protected]> Co-authored-by: Daniel (dB.) Doubrovkine <[email protected]>
austintlee · Oct 23, 2023 · 352f0a5 · 352f0a5
1 parent dbe6564
commit 352f0a5
Show file tree

Hide file tree

Showing 4 changed files with 215 additions and 11 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -143,6 +143,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix broken backward compatibility from 2.7 for IndexSorted field indices ([#10045](https://github.com/opensearch-project/OpenSearch/pull/10045))
 - Fix concurrent search NPE when track_total_hits, terminate_after and size=0 are used ([#10082](https://github.com/opensearch-project/OpenSearch/pull/10082))
 - Fix remove ingest processor handing ignore_missing parameter not correctly ([10089](https://github.com/opensearch-project/OpenSearch/pull/10089))
+- Fix class_cast_exception when passing int to _version and other metadata fields in ingest simulate API ([#10101](https://github.com/opensearch-project/OpenSearch/pull/10101))
 - Fix circular dependency in Settings initialization ([10194](https://github.com/opensearch-project/OpenSearch/pull/10194))
 - Fix registration and initialization of multiple extensions ([10256](https://github.com/opensearch-project/OpenSearch/pull/10256))
 

diff --git a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/90_simulate.yml b/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/90_simulate.yml
@@ -976,3 +976,140 @@ teardown:
           }
   - match: { error.root_cause.0.type: "illegal_argument_exception" }
   - match: { error.root_cause.0.reason: "Pipeline processor configured for non-existent pipeline [____pipeline_doesnot_exist___]" }
+
+---
+"Test simulate with docs containing metadata fields":
+  - do:
+      ingest.simulate:
+        body: >
+          {
+            "pipeline": {
+              "description": "_description",
+              "processors": [
+                {
+                  "set" : {
+                    "field": "field2",
+                    "value": "foo"
+                  }
+                }
+              ]
+            },
+            "docs": [
+              {
+                "_index": "index",
+                "_id": "id",
+                "_routing": "foo",
+                "_version": 100,
+                "_if_seq_no": 12333333333333333,
+                "_if_primary_term": 1,
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }
+
+  - length: { docs: 1 }
+  - match: { docs.0.doc._index: "index" }
+  - match: { docs.0.doc._id: "id" }
+  - match: { docs.0.doc._routing: "foo" }
+  - match: { docs.0.doc._version: "100" }
+  - match: { docs.0.doc._if_seq_no: "12333333333333333" }
+  - match: { docs.0.doc._if_primary_term: "1" }
+  - match: { docs.0.doc._source.foo: "bar" }
+
+  - do:
+      catch: bad_request
+      ingest.simulate:
+        body: >
+          {
+            "pipeline": {
+              "description": "_description",
+              "processors": [
+                {
+                  "set" : {
+                    "field" : "field2",
+                    "value": "foo"
+                  }
+                }
+              ]
+            },
+            "docs": [
+              {
+                "_index": "index",
+                "_id": "id",
+                "_routing": "foo",
+                "_version": "bar",
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }
+  - match: { status: 400 }
+  - match: { error.root_cause.0.type: "illegal_argument_exception" }
+  - match: { error.root_cause.0.reason: "Failed to parse parameter [_version], only int or long is accepted" }
+
+  - do:
+      catch: bad_request
+      ingest.simulate:
+        body: >
+          {
+            "pipeline": {
+              "description": "_description",
+              "processors": [
+                {
+                  "set" : {
+                    "field" : "field2",
+                    "value": "foo"
+                  }
+                }
+              ]
+            },
+            "docs": [
+              {
+                "_index": "index",
+                "_id": "id",
+                "_routing": "foo",
+                "_if_seq_no": "123",
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }
+  - match: { status: 400 }
+  - match: { error.root_cause.0.type: "illegal_argument_exception" }
+  - match: { error.root_cause.0.reason: "Failed to parse parameter [_if_seq_no], only int or long is accepted" }
+
+  - do:
+      catch: bad_request
+      ingest.simulate:
+        body: >
+          {
+            "pipeline": {
+              "description": "_description",
+              "processors": [
+                {
+                  "set" : {
+                    "field" : "field2",
+                    "value": "foo"
+                  }
+                }
+              ]
+            },
+            "docs": [
+              {
+                "_index": "index",
+                "_id": "id",
+                "_routing": "foo",
+                "_if_primary_term": "1",
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }
+  - match: { status: 400 }
+  - match: { error.root_cause.0.type: "illegal_argument_exception" }
+  - match: { error.root_cause.0.reason: "Failed to parse parameter [_if_primary_term], only int or long is accepted" }
diff --git a/server/src/main/java/org/opensearch/action/ingest/SimulatePipelineRequest.java b/server/src/main/java/org/opensearch/action/ingest/SimulatePipelineRequest.java
@@ -218,7 +218,12 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
             String routing = ConfigurationUtils.readOptionalStringOrIntProperty(null, null, dataMap, Metadata.ROUTING.getFieldName());
             Long version = null;
             if (dataMap.containsKey(Metadata.VERSION.getFieldName())) {
-                version = (Long) ConfigurationUtils.readObject(null, null, dataMap, Metadata.VERSION.getFieldName());
+                Object versionFieldValue = ConfigurationUtils.readObject(null, null, dataMap, Metadata.VERSION.getFieldName());
+                if (versionFieldValue instanceof Integer || versionFieldValue instanceof Long) {
+                    version = ((Number) versionFieldValue).longValue();
+                } else {
+                    throw new IllegalArgumentException("Failed to parse parameter [_version], only int or long is accepted");
+                }
             }
             VersionType versionType = null;
             if (dataMap.containsKey(Metadata.VERSION_TYPE.getFieldName())) {
@@ -228,12 +233,25 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
             }
             IngestDocument ingestDocument = new IngestDocument(index, id, routing, version, versionType, document);
             if (dataMap.containsKey(Metadata.IF_SEQ_NO.getFieldName())) {
-                Long ifSeqNo = (Long) ConfigurationUtils.readObject(null, null, dataMap, Metadata.IF_SEQ_NO.getFieldName());
-                ingestDocument.setFieldValue(Metadata.IF_SEQ_NO.getFieldName(), ifSeqNo);
+                Object ifSeqNoFieldValue = ConfigurationUtils.readObject(null, null, dataMap, Metadata.IF_SEQ_NO.getFieldName());
+                if (ifSeqNoFieldValue instanceof Integer || ifSeqNoFieldValue instanceof Long) {
+                    ingestDocument.setFieldValue(Metadata.IF_SEQ_NO.getFieldName(), ((Number) ifSeqNoFieldValue).longValue());
+                } else {
+                    throw new IllegalArgumentException("Failed to parse parameter [_if_seq_no], only int or long is accepted");
+                }
             }
             if (dataMap.containsKey(Metadata.IF_PRIMARY_TERM.getFieldName())) {
-                Long ifPrimaryTerm = (Long) ConfigurationUtils.readObject(null, null, dataMap, Metadata.IF_PRIMARY_TERM.getFieldName());
-                ingestDocument.setFieldValue(Metadata.IF_PRIMARY_TERM.getFieldName(), ifPrimaryTerm);
+                Object ifPrimaryTermFieldValue = ConfigurationUtils.readObject(
+                    null,
+                    null,
+                    dataMap,
+                    Metadata.IF_PRIMARY_TERM.getFieldName()
+                );
+                if (ifPrimaryTermFieldValue instanceof Integer || ifPrimaryTermFieldValue instanceof Long) {
+                    ingestDocument.setFieldValue(Metadata.IF_PRIMARY_TERM.getFieldName(), ((Number) ifPrimaryTermFieldValue).longValue());
+                } else {
+                    throw new IllegalArgumentException("Failed to parse parameter [_if_primary_term], only int or long is accepted");
+                }
             }
             ingestDocumentList.add(ingestDocument);
         }

diff --git a/server/src/test/java/org/opensearch/action/ingest/SimulatePipelineRequestParsingTests.java b/server/src/test/java/org/opensearch/action/ingest/SimulatePipelineRequestParsingTests.java
@@ -144,17 +144,29 @@ public void innerTestParseWithProvidedPipeline() throws Exception {
             List<IngestDocument.Metadata> fields = Arrays.asList(INDEX, ID, ROUTING, VERSION, VERSION_TYPE, IF_SEQ_NO, IF_PRIMARY_TERM);
             for (IngestDocument.Metadata field : fields) {
                 if (field == VERSION) {
-                    Long value = randomLong();
-                    doc.put(field.getFieldName(), value);
-                    expectedDoc.put(field.getFieldName(), value);
+                    if (randomBoolean()) {
+                        Long value = randomLong();
+                        doc.put(field.getFieldName(), value);
+                        expectedDoc.put(field.getFieldName(), value);
+                    } else {
+                        Integer value = randomIntBetween(1, 1000000);
+                        doc.put(field.getFieldName(), value);
+                        expectedDoc.put(field.getFieldName(), value);
+                    }
                 } else if (field == VERSION_TYPE) {
                     String value = VersionType.toString(randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL, VersionType.EXTERNAL_GTE));
                     doc.put(field.getFieldName(), value);
                     expectedDoc.put(field.getFieldName(), value);
                 } else if (field == IF_SEQ_NO || field == IF_PRIMARY_TERM) {
-                    Long value = randomNonNegativeLong();
-                    doc.put(field.getFieldName(), value);
-                    expectedDoc.put(field.getFieldName(), value);
+                    if (randomBoolean()) {
+                        Long value = randomNonNegativeLong();
+                        doc.put(field.getFieldName(), value);
+                        expectedDoc.put(field.getFieldName(), value);
+                    } else {
+                        Integer value = randomIntBetween(1, 1000000);
+                        doc.put(field.getFieldName(), value);
+                        expectedDoc.put(field.getFieldName(), value);
+                    }
                 } else {
                     if (randomBoolean()) {
                         String value = randomAlphaOfLengthBetween(1, 10);
@@ -282,4 +294,40 @@ public void testNotValidDocs() {
         );
         assertThat(e3.getMessage(), containsString("required property is missing"));
     }
+
+    public void testNotValidMetadataFields() {
+        List<IngestDocument.Metadata> fields = Arrays.asList(VERSION, IF_SEQ_NO, IF_PRIMARY_TERM);
+        for (IngestDocument.Metadata field : fields) {
+            String metadataFieldName = field.getFieldName();
+            Map<String, Object> requestContent = new HashMap<>();
+            List<Map<String, Object>> docs = new ArrayList<>();
+            requestContent.put(Fields.DOCS, docs);
+            Map<String, Object> doc = new HashMap<>();
+            doc.put(metadataFieldName, randomAlphaOfLengthBetween(1, 10));
+            doc.put(Fields.SOURCE, Collections.singletonMap(randomAlphaOfLengthBetween(1, 10), randomAlphaOfLengthBetween(1, 10)));
+            docs.add(doc);
+
+            Map<String, Object> pipelineConfig = new HashMap<>();
+            List<Map<String, Object>> processors = new ArrayList<>();
+            Map<String, Object> processorConfig = new HashMap<>();
+            List<Map<String, Object>> onFailureProcessors = new ArrayList<>();
+            int numOnFailureProcessors = randomIntBetween(0, 1);
+            for (int j = 0; j < numOnFailureProcessors; j++) {
+                onFailureProcessors.add(Collections.singletonMap("mock_processor", Collections.emptyMap()));
+            }
+            if (numOnFailureProcessors > 0) {
+                processorConfig.put("on_failure", onFailureProcessors);
+            }
+            processors.add(Collections.singletonMap("mock_processor", processorConfig));
+            pipelineConfig.put("processors", processors);
+
+            requestContent.put(Fields.PIPELINE, pipelineConfig);
+
+            assertThrows(
+                "Failed to parse parameter [" + metadataFieldName + "], only int or long is accepted",
+                IllegalArgumentException.class,
+                () -> SimulatePipelineRequest.parse(requestContent, false, ingestService)
+            );
+        }
+    }
 }