fix index out of bounds error in KV Processor (#22288)

- checks for index-out-of-bounds
- added unit tests for failed `field_split` and `value_split` scenarios

missed this test in #22272.

Author: talevy

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/KeyValueProcessor.java

@@ -98,7 +98,13 @@ public void execute(IngestDocument document) {
 
         String fieldPathPrefix = (targetField == null) ? "" : targetField + ".";
         Arrays.stream(oldVal.split(fieldSplit))
-            .map((f) -> f.split(valueSplit, 2))
+            .map((f) -> {
+                String[] kv = f.split(valueSplit, 2);
+                if (kv.length != 2) {
+                    throw new IllegalArgumentException("field [" + field + "] does not contain value_split [" + valueSplit + "]");
+                }
+                return kv;
+            })
             .filter((p) -> includeKeys == null || includeKeys.contains(p[0]))
             .forEach((p) -> append(document, fieldPathPrefix + p[0], p[1]));
     }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/KeyValueProcessorTests.java

@@ -93,4 +93,20 @@ public void testNonExistentWithIgnoreMissing() throws Exception {
         processor.execute(ingestDocument);
         assertIngestDocument(originalIngestDocument, ingestDocument);
     }
+
+    public void testFailFieldSplitMatch() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "first=hello|second=world|second=universe");
+        Processor processor = new KeyValueProcessor(randomAsciiOfLength(10), fieldName, "&", "=", null, "target", false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue("target.first", String.class), equalTo("hello|second=world|second=universe"));
+        assertFalse(ingestDocument.hasField("target.second"));
+    }
+
+    public void testFailValueSplitMatch() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.singletonMap("foo", "bar"));
+        Processor processor = new KeyValueProcessor(randomAsciiOfLength(10), "foo", "&", "=", null, "target", false);
+        Exception exception = expectThrows(IllegalArgumentException.class, () -> processor.execute(ingestDocument));
+        assertThat(exception.getMessage(), equalTo("field [foo] does not contain value_split [=]"));
+    }
 }