Add a test to ensure shard failures do not redirect when node feature is missing.

Author: jbaiera

server/src/test/java/org/elasticsearch/action/bulk/BulkOperationTests.java

@@ -441,6 +441,45 @@ public void testFailingDocumentRedirectsToFailureStore() throws Exception {
         assertThat(failedItem.getFailureStoreStatus(), equalTo(IndexDocFailureStoreStatus.USED));
     }
 
+    /**
+     * A bulk operation to a data stream with a failure store enabled should NOT redirect any documents that fail at a shard level to the
+     * failure store if the failure store node feature is not on every node in the cluster
+     */
+    public void testFailingDocumentIgnoredByFailureStoreWhenFeatureIsDisabled() throws Exception {
+        Assume.assumeTrue(DataStream.isFailureStoreFeatureFlagEnabled());
+
+        // Requests that go to two separate shards
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest(fsDataStreamName).id("1").source(Map.of("key", "val")).opType(DocWriteRequest.OpType.CREATE));
+        bulkRequest.add(new IndexRequest(fsDataStreamName).id("3").source(Map.of("key", "val")).opType(DocWriteRequest.OpType.CREATE));
+
+        NodeClient client = getNodeClient(
+            thatFailsDocuments(Map.of(new IndexAndId(ds2BackingIndex1.getIndex().getName(), "3"), () -> new MapperException("test")))
+        );
+
+        BulkResponse bulkItemResponses = safeAwait(
+            l -> newBulkOperation(
+                clusterState,
+                client,
+                bulkRequest,
+                new AtomicArray<>(bulkRequest.numberOfActions()),
+                mockObserver(clusterState),
+                l,
+                new FailureStoreDocumentConverter(),
+                DataStreamFailureStoreSettings.create(ClusterSettings.createBuiltInClusterSettings()),
+                false
+            ).run()
+        );
+        assertThat(bulkItemResponses.hasFailures(), is(true));
+        BulkItemResponse failedItem = Arrays.stream(bulkItemResponses.getItems())
+            .filter(BulkItemResponse::isFailed)
+            .findFirst()
+            .orElseThrow(() -> new AssertionError("Could not find redirected item"));
+        assertThat(failedItem.getFailure().getCause(), is(instanceOf(MapperException.class)));
+        assertThat(failedItem.getFailure().getCause().getMessage(), is(equalTo("test")));
+        assertThat(failedItem.getFailureStoreStatus(), equalTo(IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN));
+    }
+
     public void testFailingDocumentRedirectsToFailureStoreWhenEnabledByClusterSetting() {
         Assume.assumeTrue(DataStream.isFailureStoreFeatureFlagEnabled());
 
@@ -482,7 +521,8 @@ public void testFailingDocumentRedirectsToFailureStoreWhenEnabledByClusterSettin
                 mockObserver(clusterState),
                 l,
                 new FailureStoreDocumentConverter(),
-                dataStreamFailureStoreSettings
+                dataStreamFailureStoreSettings,
+                true
             ).run()
         );
         assertThat(bulkItemResponsesUsingClusterSetting.hasFailures(), is(false));
@@ -1175,7 +1215,8 @@ private BulkOperation newBulkOperation(
             observer,
             listener,
             failureStoreDocumentConverter,
-            DataStreamFailureStoreSettings.create(ClusterSettings.createBuiltInClusterSettings())
+            DataStreamFailureStoreSettings.create(ClusterSettings.createBuiltInClusterSettings()),
+            true
         );
     }
 
@@ -1187,7 +1228,8 @@ private BulkOperation newBulkOperation(
         ClusterStateObserver observer,
         ActionListener<BulkResponse> listener,
         FailureStoreDocumentConverter failureStoreDocumentConverter,
-        DataStreamFailureStoreSettings dataStreamFailureStoreSettings
+        DataStreamFailureStoreSettings dataStreamFailureStoreSettings,
+        boolean failureStoreNodeFeatureEnabled
     ) {
         // Time provision
         long timeZero = TimeUnit.MILLISECONDS.toNanos(randomMillisUpToYear9999() - TimeUnit.DAYS.toMillis(1));
@@ -1222,7 +1264,7 @@ private BulkOperation newBulkOperation(
             failureStoreDocumentConverter,
             FailureStoreMetrics.NOOP,
             dataStreamFailureStoreSettings,
-            true
+            failureStoreNodeFeatureEnabled
         );
     }