Fix long handling for version, if_seq_no and if_primary_term (#69238)

After #66197 we've lost ability to use longs for version, if_seq_no and if_primary_term (only ints would work).
That slipped through because 1 test went missing after refactoring in #50131.
This change fixes that with new ConfigurationUtils.readOptionalStringOrLongProperty method and and brings back the test.

Author: probakowski

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

@@ -175,7 +175,7 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
                 dataMap, Metadata.ROUTING.getFieldName());
             Long version = null;
             if (dataMap.containsKey(Metadata.VERSION.getFieldName())) {
-                String versionValue = ConfigurationUtils.readOptionalStringOrIntProperty(null, null,
+                String versionValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
                     dataMap, Metadata.VERSION.getFieldName());
                 if (versionValue != null) {
                     version = Long.valueOf(versionValue);
@@ -191,7 +191,7 @@ 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())) {
-                String ifSeqNoValue = ConfigurationUtils.readOptionalStringOrIntProperty(null, null,
+                String ifSeqNoValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
                     dataMap, Metadata.IF_SEQ_NO.getFieldName());
                 if (ifSeqNoValue != null) {
                     Long ifSeqNo = Long.valueOf(ifSeqNoValue);
@@ -201,7 +201,7 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
                 }
             }
             if (dataMap.containsKey(Metadata.IF_PRIMARY_TERM.getFieldName())) {
-                String ifPrimaryTermValue = ConfigurationUtils.readOptionalStringOrIntProperty(null, null,
+                String ifPrimaryTermValue = ConfigurationUtils.readOptionalStringOrLongProperty(null, null,
                     dataMap, Metadata.IF_PRIMARY_TERM.getFieldName());
                 if (ifPrimaryTermValue != null) {
                     Long ifPrimaryTerm = Long.valueOf(ifPrimaryTermValue);

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

@@ -101,7 +101,7 @@ public void testParseUsingPipelineStore() throws Exception {
         assertThat(actualRequest.getPipeline().getProcessors().size(), equalTo(1));
     }
 
-    public void innerTestParseWithProvidedPipeline() throws Exception {
+    public void testParseWithProvidedPipeline() throws Exception {
         int numDocs = randomIntBetween(1, 10);
 
         Map<String, Object> requestContent = new HashMap<>();
@@ -113,12 +113,22 @@ public void innerTestParseWithProvidedPipeline() throws Exception {
             Map<String, Object> expectedDoc = new HashMap<>();
             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_TYPE) {
+                if (field == VERSION) {
+                    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) {
+                    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 (randomBoolean()) {
                         String value = randomAlphaOfLengthBetween(1, 10);