ReplicationOperation exceptions should not escape (#116074)

We introduce ActionListener.run() in order to ensure the
RefCountingListener introduced by PR #115341 , is the single point
that is failed upon exceptions, and no exception escapes through
the ReplicationOperation.execute() method.

Fixes #116071
Fixes #116081
Fixes #116073

Author: kingherc

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

@@ -115,28 +115,25 @@ public void execute() throws Exception {
             );
             resultListener.onResponse(primaryResult);
         }, resultListener::onFailure))) {
-            final String activeShardCountFailure = checkActiveShardCount();
-            final ShardRouting primaryRouting = primary.routingEntry();
-            final ShardId primaryId = primaryRouting.shardId();
-            if (activeShardCountFailure != null) {
-                pendingActionsListener.acquire()
-                    .onFailure(
-                        new UnavailableShardsException(
-                            primaryId,
-                            "{} Timeout: [{}], request: [{}]",
-                            activeShardCountFailure,
-                            request.timeout(),
-                            request
-                        )
+            ActionListener.run(pendingActionsListener.acquire(), (primaryCoordinationListener) -> { // triggered when we finish coordination
+                final String activeShardCountFailure = checkActiveShardCount();
+                final ShardRouting primaryRouting = primary.routingEntry();
+                final ShardId primaryId = primaryRouting.shardId();
+                if (activeShardCountFailure != null) {
+                    throw new UnavailableShardsException(
+                        primaryId,
+                        "{} Timeout: [{}], request: [{}]",
+                        activeShardCountFailure,
+                        request.timeout(),
+                        request
                     );
-                return;
-            }
+                }
 
-            totalShards.incrementAndGet();
-            var primaryCoordinationPendingActionListener = pendingActionsListener.acquire(); // triggered when we finish all coordination
-            primary.perform(request, primaryCoordinationPendingActionListener.delegateFailureAndWrap((l, primaryResult) -> {
-                handlePrimaryResult(primaryResult, l, pendingActionsListener);
-            }));
+                totalShards.incrementAndGet();
+                primary.perform(request, primaryCoordinationListener.delegateFailureAndWrap((l, primaryResult) -> {
+                    handlePrimaryResult(primaryResult, l, pendingActionsListener);
+                }));
+            });
         }
     }
 
@@ -153,24 +150,25 @@ private void handlePrimaryResult(
             }
             final ReplicationGroup replicationGroup = primary.getReplicationGroup();
 
-            var primaryOperationPendingActionListener = pendingActionsListener.acquire();
-            replicasProxy.onPrimaryOperationComplete(
-                replicaRequest,
-                replicationGroup.getRoutingTable(),
-                ActionListener.wrap(ignored -> primaryOperationPendingActionListener.onResponse(null), exception -> {
-                    totalShards.incrementAndGet();
-                    shardReplicaFailures.add(
-                        new ReplicationResponse.ShardInfo.Failure(
-                            primary.routingEntry().shardId(),
-                            null,
-                            exception,
-                            ExceptionsHelper.status(exception),
-                            false
-                        )
-                    );
-                    primaryOperationPendingActionListener.onResponse(null);
-                })
-            );
+            ActionListener.run(pendingActionsListener.acquire(), primaryOperationPendingActionListener -> {
+                replicasProxy.onPrimaryOperationComplete(
+                    replicaRequest,
+                    replicationGroup.getRoutingTable(),
+                    ActionListener.wrap(ignored -> primaryOperationPendingActionListener.onResponse(null), exception -> {
+                        totalShards.incrementAndGet();
+                        shardReplicaFailures.add(
+                            new ReplicationResponse.ShardInfo.Failure(
+                                primary.routingEntry().shardId(),
+                                null,
+                                exception,
+                                ExceptionsHelper.status(exception),
+                                false
+                            )
+                        );
+                        primaryOperationPendingActionListener.onResponse(null);
+                    })
+                );
+            });
 
             // we have to get the replication group after successfully indexing into the primary in order to honour recovery semantics.
             // we have to make sure that every operation indexed into the primary after recovery start will also be replicated
@@ -234,13 +232,14 @@ private void markUnavailableShardsAsStale(
     ) {
         // if inSyncAllocationIds contains allocation ids of shards that don't exist in RoutingTable, mark copies as stale
         for (String allocationId : replicationGroup.getUnavailableInSyncShards()) {
-            var staleCopyPendingActionListener = pendingActionsListener.acquire();
-            replicasProxy.markShardCopyAsStaleIfNeeded(
-                replicaRequest.shardId(),
-                allocationId,
-                primaryTerm,
-                staleCopyPendingActionListener.delegateResponse((l, e) -> onNoLongerPrimary(e, l))
-            );
+            ActionListener.run(pendingActionsListener.acquire(), (staleCopyPendingActionListener) -> {
+                replicasProxy.markShardCopyAsStaleIfNeeded(
+                    replicaRequest.shardId(),
+                    allocationId,
+                    primaryTerm,
+                    staleCopyPendingActionListener.delegateResponse((l, e) -> onNoLongerPrimary(e, l))
+                );
+            });
         }
     }
 
@@ -285,13 +284,17 @@ private void performOnReplica(
             logger.trace("[{}] sending op [{}] to replica {} for request [{}]", shard.shardId(), opType, shard, replicaRequest);
         }
         totalShards.incrementAndGet();
-        var replicationPendingActionListener = pendingActionsListener.acquire();
-        ActionListener.run(replicationPendingActionListener, (listener) -> {
+        ActionListener.run(pendingActionsListener.acquire(), (replicationPendingActionListener) -> {
             final ActionListener<ReplicaResponse> replicationListener = new ActionListener<>() {
                 @Override
                 public void onResponse(ReplicaResponse response) {
                     successfulShards.incrementAndGet();
-                    updateCheckPoints(shard, response::localCheckpoint, response::globalCheckpoint, () -> listener.onResponse(null));
+                    updateCheckPoints(
+                        shard,
+                        response::localCheckpoint,
+                        response::globalCheckpoint,
+                        () -> replicationPendingActionListener.onResponse(null)
+                    );
                 }
 
                 @Override
@@ -325,7 +328,7 @@ public void onFailure(Exception replicaException) {
                         primaryTerm,
                         message,
                         replicaException,
-                        listener.delegateResponse((l, e) -> onNoLongerPrimary(e, l))
+                        replicationPendingActionListener.delegateResponse((l, e) -> onNoLongerPrimary(e, l))
                     );
                 }