Fix ArrayIndexOutOfBoundsException in ShardBulkInferenceActionFilter (elastic#122538)

(cherry picked from commit 229d392)

# Conflicts:
#	x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilterIT.java

Author: Mikep86

docs/changelog/122538.yaml

@@ -0,0 +1,5 @@
+pr: 122538
+summary: Fix `ArrayIndexOutOfBoundsException` in `ShardBulkInferenceActionFilter`
+area: Ingest
+type: bug
+issues: []

x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilterIT.java

@@ -37,6 +37,7 @@
 import java.util.Set;
 
 import static org.elasticsearch.xpack.inference.mapper.SemanticTextFieldTests.randomSemanticTextInput;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class ShardBulkInferenceActionFilterIT extends ESIntegTestCase {
@@ -152,4 +153,47 @@ public void testBulkOperations() throws Exception {
         }
     }
 
+    public void testItemFailures() {
+        prepareCreate(INDEX_NAME).setMapping(
+            String.format(
+                Locale.ROOT,
+                """
+                    {
+                        "properties": {
+                            "sparse_field": {
+                                "type": "semantic_text",
+                                "inference_id": "%s"
+                            },
+                            "dense_field": {
+                                "type": "semantic_text",
+                                "inference_id": "%s"
+                            }
+                        }
+                    }
+                    """,
+                TestSparseInferenceServiceExtension.TestInferenceService.NAME,
+                TestDenseInferenceServiceExtension.TestInferenceService.NAME
+            )
+        ).get();
+
+        BulkRequestBuilder bulkReqBuilder = client().prepareBulk();
+        int totalBulkSize = randomIntBetween(100, 200);  // Use a bulk request size large enough to require batching
+        for (int bulkSize = 0; bulkSize < totalBulkSize; bulkSize++) {
+            String id = Integer.toString(bulkSize);
+
+            // Set field values that will cause errors when generating inference requests
+            Map<String, Object> source = new HashMap<>();
+            source.put("sparse_field", List.of(Map.of("foo", "bar"), Map.of("baz", "bar")));
+            source.put("dense_field", List.of(Map.of("foo", "bar"), Map.of("baz", "bar")));
+
+            bulkReqBuilder.add(new IndexRequestBuilder(client()).setIndex(INDEX_NAME).setId(id).setSource(source));
+        }
+
+        BulkResponse bulkResponse = bulkReqBuilder.get();
+        assertThat(bulkResponse.hasFailures(), equalTo(true));
+        for (BulkItemResponse bulkItemResponse : bulkResponse.getItems()) {
+            assertThat(bulkItemResponse.isFailed(), equalTo(true));
+            assertThat(bulkItemResponse.getFailureMessage(), containsString("expected [String|Number|Boolean]"));
+        }
+    }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilter.java

@@ -427,7 +427,7 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu
                     isUpdateRequest = true;
                     if (updateRequest.script() != null) {
                         addInferenceResponseFailure(
-                            item.id(),
+                            itemIndex,
                             new ElasticsearchStatusException(
                                 "Cannot apply update with a script on indices that contain [{}] field(s)",
                                 RestStatus.BAD_REQUEST,
@@ -458,7 +458,7 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu
                         if (valueObj == null) {
                             if (isUpdateRequest) {
                                 addInferenceResponseFailure(
-                                    item.id(),
+                                    itemIndex,
                                     new ElasticsearchStatusException(
                                         "Field [{}] must be specified on an update request to calculate inference for field [{}]",
                                         RestStatus.BAD_REQUEST,
@@ -475,7 +475,7 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu
                         try {
                             values = nodeStringValues(field, valueObj);
                         } catch (Exception exc) {
-                            addInferenceResponseFailure(item.id(), exc);
+                            addInferenceResponseFailure(itemIndex, exc);
                             break;
                         }
                         List<FieldInferenceRequest> fieldRequests = fieldRequestsMap.computeIfAbsent(inferenceId, k -> new ArrayList<>());