Fix bulk NPE when retrying failure redirect after cluster block (elas…

…tic#107598) (elastic#107793)

This PR fixes a bug in the bulk operation when retrying blocked cluster states before 
executing a failure store write by correctly wrapping the retry runnable to keep it from 
prematurely returning a null response.

docs/changelog/107598.yaml

@@ -0,0 +1,5 @@
+pr: 107598
+summary: Fix bulk NPE when retrying failure redirect after cluster block
+area: Data streams
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java

@@ -161,7 +161,11 @@ private void doRedirectFailures() {
         assert failureStoreRedirects.isEmpty() != true : "Attempting to redirect failures, but none were present in the queue";
         final ClusterState clusterState = observer.setAndGetObservedState();
         // If the cluster is blocked at this point, discard the failure store redirects and complete the response with the original failures
-        if (handleBlockExceptions(clusterState, ActionRunnable.run(listener, this::doRedirectFailures), this::discardRedirectsAndFinish)) {
+        if (handleBlockExceptions(
+            clusterState,
+            ActionRunnable.wrap(listener, (l) -> this.doRedirectFailures()),
+            this::discardRedirectsAndFinish
+        )) {
             return;
         }
         Map<ShardId, List<BulkItemRequest>> requestsByShard = drainAndGroupRedirectsByShards(clusterState);