Fork to WRITE thread before failing shard in updateCheckPoints (elastic#87458) (elastic#87496)

Failing a shard may block on IO so must not happen on a transport worker
thread. With this commit we use a `WRITE` thread to handle shard
failures caused by exceptions thrown within `updateCheckPoints`.

Closes elastic#87094

Author: DaveCTurner

docs/changelog/87458.yaml

@@ -0,0 +1,6 @@
+pr: 87458
+summary: Fork to WRITE thread before failing shard in `updateCheckPoints`
+area: Engine
+type: bug
+issues:
+ - 87094

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -155,11 +155,12 @@ private void handlePrimaryResult(final PrimaryResultT primaryResult) {
             @Override
             public void onResponse(Void aVoid) {
                 successfulShards.incrementAndGet();
-                try {
-                    updateCheckPoints(primary.routingEntry(), primary::localCheckpoint, primary::globalCheckpoint);
-                } finally {
-                    decPendingAndFinishIfNeeded();
-                }
+                updateCheckPoints(
+                    primary.routingEntry(),
+                    primary::localCheckpoint,
+                    primary::globalCheckpoint,
+                    () -> decPendingAndFinishIfNeeded()
+                );
             }
 
             @Override
@@ -221,11 +222,7 @@ private void performOnReplica(
             @Override
             public void onResponse(ReplicaResponse response) {
                 successfulShards.incrementAndGet();
-                try {
-                    updateCheckPoints(shard, response::localCheckpoint, response::globalCheckpoint);
-                } finally {
-                    decPendingAndFinishIfNeeded();
-                }
+                updateCheckPoints(shard, response::localCheckpoint, response::globalCheckpoint, () -> decPendingAndFinishIfNeeded());
             }
 
             @Override
@@ -302,16 +299,46 @@ public boolean shouldRetry(Exception e) {
         replicationAction.run();
     }
 
-    private void updateCheckPoints(ShardRouting shard, LongSupplier localCheckpointSupplier, LongSupplier globalCheckpointSupplier) {
+    private void updateCheckPoints(
+        ShardRouting shard,
+        LongSupplier localCheckpointSupplier,
+        LongSupplier globalCheckpointSupplier,
+        Runnable onCompletion
+    ) {
+        boolean forked = false;
         try {
             primary.updateLocalCheckpointForShard(shard.allocationId().getId(), localCheckpointSupplier.getAsLong());
             primary.updateGlobalCheckpointForShard(shard.allocationId().getId(), globalCheckpointSupplier.getAsLong());
         } catch (final AlreadyClosedException e) {
             // the index was deleted or this shard was never activated after a relocation; fall through and finish normally
         } catch (final Exception e) {
-            // fail the primary but fall through and let the rest of operation processing complete
-            final String message = String.format(Locale.ROOT, "primary failed updating local checkpoint for replica %s", shard);
-            primary.failShard(message, e);
+            threadPool.executor(ThreadPool.Names.WRITE).execute(new AbstractRunnable() {
+                @Override
+                public void onFailure(Exception e) {
+                    assert false : e;
+                }
+
+                @Override
+                public boolean isForceExecution() {
+                    return true;
+                }
+
+                @Override
+                protected void doRun() {
+                    // fail the primary but fall through and let the rest of operation processing complete
+                    primary.failShard(String.format(Locale.ROOT, "primary failed updating local checkpoint for replica %s", shard), e);
+                }
+
+                @Override
+                public void onAfter() {
+                    onCompletion.run();
+                }
+            });
+            forked = true;
+        } finally {
+            if (forked == false) {
+                onCompletion.run();
+            }
         }
     }
 

server/src/test/java/org/elasticsearch/action/support/replication/ReplicationOperationTests.java

@@ -510,8 +510,9 @@ public void testPrimaryFailureHandlingReplicaResponse() throws Exception {
         final Set<String> trackedShards = shardRoutingTable.getAllAllocationIds();
         final ReplicationGroup initialReplicationGroup = new ReplicationGroup(shardRoutingTable, inSyncAllocationIds, trackedShards, 0);
 
+        final Thread testThread = Thread.currentThread();
         final boolean fatal = randomBoolean();
-        final AtomicBoolean primaryFailed = new AtomicBoolean();
+        final PlainActionFuture<Void> primaryFailedFuture = new PlainActionFuture<>();
         final ReplicationOperation.Primary<Request, Request, TestPrimary.Result> primary = new TestPrimary(
             primaryRouting,
             () -> initialReplicationGroup,
@@ -520,7 +521,10 @@ public void testPrimaryFailureHandlingReplicaResponse() throws Exception {
 
             @Override
             public void failShard(String message, Exception exception) {
-                primaryFailed.set(true);
+                assertNotSame(testThread, Thread.currentThread());
+                assertThat(Thread.currentThread().getName(), containsString('[' + ThreadPool.Names.WRITE + ']'));
+                assertTrue(fatal);
+                primaryFailedFuture.onResponse(null);
             }
 
             @Override
@@ -543,7 +547,10 @@ public void updateLocalCheckpointForShard(String allocationId, long checkpoint)
         TestReplicationOperation operation = new TestReplicationOperation(request, primary, listener, replicas, primaryTerm);
         operation.execute();
 
-        assertThat(primaryFailed.get(), equalTo(fatal));
+        if (fatal) {
+            primaryFailedFuture.get(10, TimeUnit.SECONDS);
+        }
+
         final ShardInfo shardInfo = listener.actionGet().getShardInfo();
         assertThat(shardInfo.getFailed(), equalTo(0));
         assertThat(shardInfo.getFailures(), arrayWithSize(0));