[7.x] Accept more ingest simulate params as ints or strings (#66197, #69238) (#69262)

Author: probakowski

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

@@ -184,7 +184,13 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
                 dataMap, Metadata.ROUTING.getFieldName());
             Long version = null;
             if (dataMap.containsKey(Metadata.VERSION.getFieldName())) {
-                version = (Long) ConfigurationUtils.readObject(null, null, dataMap, Metadata.VERSION.getFieldName());
+                String versionValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
+                    dataMap, Metadata.VERSION.getFieldName());
+                if (versionValue != null) {
+                    version = Long.valueOf(versionValue);
+                } else {
+                    throw new IllegalArgumentException("[_version] cannot be null");
+                }
             }
             VersionType versionType = null;
             if (dataMap.containsKey(Metadata.VERSION_TYPE.getFieldName())) {
@@ -194,12 +200,24 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
             IngestDocument ingestDocument =
                 new IngestDocument(index, type, 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());
+                String ifSeqNoValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
+                    dataMap, Metadata.IF_SEQ_NO.getFieldName());
+                if (ifSeqNoValue != null) {
+                    Long ifSeqNo = Long.valueOf(ifSeqNoValue);
                 ingestDocument.setFieldValue(Metadata.IF_SEQ_NO.getFieldName(), ifSeqNo);
+                } else {
+                    throw new IllegalArgumentException("[_if_seq_no] cannot be null");
+                }
             }
             if (dataMap.containsKey(Metadata.IF_PRIMARY_TERM.getFieldName())) {
-                Long ifPrimaryTerm = (Long) ConfigurationUtils.readObject(null, null, dataMap, Metadata.IF_PRIMARY_TERM.getFieldName());
+                String ifPrimaryTermValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
+                    dataMap, Metadata.IF_PRIMARY_TERM.getFieldName());
+                if (ifPrimaryTermValue != null) {
+                    Long ifPrimaryTerm = Long.valueOf(ifPrimaryTermValue);
                 ingestDocument.setFieldValue(Metadata.IF_PRIMARY_TERM.getFieldName(), ifPrimaryTerm);
+                } else {
+                    throw new IllegalArgumentException("[_if_primary_term] cannot be null");
+                }
             }
             ingestDocumentList.add(ingestDocument);
         }

server/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java

@@ -139,6 +139,34 @@ public static String readOptionalStringOrIntProperty(String processorType, Strin
         return readStringOrInt(processorType, processorTag, propertyName, value);
     }
 
+    private static String readStringOrLong(String processorType, String processorTag,
+                                           String propertyName, Object value) {
+        if (value == null) {
+            return null;
+        }
+        if (value instanceof String) {
+            return (String) value;
+        } else if (value instanceof Long || value instanceof Integer) {
+            return String.valueOf(value);
+        }
+        throw newConfigurationException(processorType, processorTag, propertyName,
+            "property isn't a string or long, but of type [" + value.getClass().getName() + "]");
+    }
+
+    /**
+     * Returns and removes the specified property from the specified configuration map.
+     *
+     * If the property value isn't of type string or long a {@link ElasticsearchParseException} is thrown.
+     */
+    public static String readOptionalStringOrLongProperty(String processorType, String processorTag,
+                                                         Map<String, Object> configuration, String propertyName) {
+        Object value = configuration.remove(propertyName);
+        if (value == null) {
+            return null;
+        }
+        return readStringOrLong(processorType, processorTag, propertyName, value);
+    }
+
     public static Boolean readBooleanProperty(String processorType, String processorTag, Map<String, Object> configuration,
                                              String propertyName, boolean defaultValue) {
         Object value = configuration.remove(propertyName);

server/src/test/java/org/elasticsearch/action/ingest/SimulatePipelineRequestParsingTests.java

@@ -145,19 +145,21 @@ private void innerTestParseWithProvidedPipeline(boolean useExplicitType) throws
                 IF_PRIMARY_TERM);
             for(IngestDocument.Metadata field : fields) {
                 if (field == VERSION) {
-                    Long value = randomLong();
-                    doc.put(field.getFieldName(), value);
-                    expectedDoc.put(field.getFieldName(), value);
+                    Object value = randomBoolean() ? randomLong() : randomInt();
+                    doc.put(field.getFieldName(), randomBoolean() ? value : value.toString());
+                    long longValue = (long) value;
+                    expectedDoc.put(field.getFieldName(), longValue);
                 } 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);
+                    Object value = randomBoolean() ? randomNonNegativeLong() : randomInt(1000);
+                    doc.put(field.getFieldName(), randomBoolean() ? value : value.toString());
+                    long longValue = (long) value;
+                    expectedDoc.put(field.getFieldName(), longValue);
                 } else if (field == TYPE) {
                     if (useExplicitType) {
                         String value = randomAlphaOfLengthBetween(1, 10);