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

CHANGELOG.md

@@ -141,10 +141,11 @@ 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:/opensearch-project/OpenSearch/pull/10045))
 - Fix concurrent search NPE when track_total_hits, terminate_after and size=0 are used ([#10082](https:/opensearch-project/OpenSearch/pull/10082))
 - Fix remove ingest processor handing ignore_missing parameter not correctly ([10089](https:/opensearch-project/OpenSearch/pull/10089))
+- Fix class_cast_exception when passing int to _version and other metadata fields in ingest simulate API ([#10101](https:/opensearch-project/OpenSearch/pull/10101))
 - Fix circular dependency in Settings initialization ([10194](https:/opensearch-project/OpenSearch/pull/10194))
 - Fix registration and initialization of multiple extensions ([10256](https:/opensearch-project/OpenSearch/pull/10256))
 
 ### Security
 
 [Unreleased 3.0]: https:/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https:/opensearch-project/OpenSearch/compare/2.11...2.x
+[Unreleased 2.x]: https:/opensearch-project/OpenSearch/compare/2.11...2.x

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" }

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