Fix class_cast_exception when passing int to _version and other metadata fields in ingest simulate API

CHANGELOG.md

@@ -123,6 +123,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))
 
 ### Security
 

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/90_simulate.yml

@@ -976,3 +976,75 @@ 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",
+                "_if_seq_no": 123,
+                "_if_primary_term": 1,
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }

server/src/main/java/org/opensearch/action/ingest/SimulatePipelineRequest.java

@@ -218,7 +218,12 @@
             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 @@
             }
             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);
         }

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);