Improve scalability of BroadcastReplicationActions

docs/changelog/92729.yaml

@@ -0,0 +1,5 @@
+pr: 92729
+summary: Improve scalability of `BroadcastReplicationActions`
+area: Network
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/action/ActionListener.java

@@ -13,6 +13,8 @@
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.CheckedFunction;
 import org.elasticsearch.core.CheckedRunnable;
+import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.Releasables;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -252,6 +254,13 @@ public String toString() {
         }
     }
 
+    /**
+     * Creates a listener which releases the given resource on completion (whether success or failure)
+     */
+    static <Response> ActionListener<Response> releasing(Releasable releasable) {
+        return wrap(runnableFromReleasable(releasable));
+    }
+
     /**
      * Creates a listener that listens for a response (or failure) and executes the
      * corresponding runnable when the response (or failure) is received.
@@ -362,6 +371,14 @@ static <Response> ActionListener<Response> runAfter(ActionListener<Response> del
         return new RunAfterActionListener<>(delegate, runAfter);
     }
 
+    /**
+     * Wraps a given listener and returns a new listener which releases the provided {@code releaseAfter}
+     * resource when the listener is notified via either {@code #onResponse} or {@code #onFailure}.
+     */
+    static <Response> ActionListener<Response> releaseAfter(ActionListener<Response> delegate, Releasable releaseAfter) {
+        return new RunAfterActionListener<>(delegate, runnableFromReleasable(releaseAfter));
+    }
+
     final class RunAfterActionListener<T> extends Delegating<T, T> {
 
         private final Runnable runAfter;
@@ -498,4 +515,18 @@ static <Response> void completeWith(ActionListener<Response> listener, CheckedSu
             throw ex;
         }
     }
+
+    private static Runnable runnableFromReleasable(Releasable releasable) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                Releasables.closeExpectNoException(releasable);
+            }
+
+            @Override
+            public String toString() {
+                return "release[" + releasable + "]";
+            }
+        };
+    }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.List;
@@ -46,15 +47,11 @@ public TransportFlushAction(
             client,
             actionFilters,
             indexNameExpressionResolver,
-            TransportShardFlushAction.TYPE
+            TransportShardFlushAction.TYPE,
+            ThreadPool.Names.FLUSH
         );
     }
 
-    @Override
-    protected ReplicationResponse newShardResponse() {
-        return new ReplicationResponse();
-    }
-
     @Override
     protected ShardFlushRequest newShardRequest(FlushRequest request, ShardId shardId) {
         return new ShardFlushRequest(request, shardId);

server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -68,6 +69,11 @@ public TransportShardFlushAction(
         );
     }
 
+    @Override
+    protected ClusterBlockLevel globalBlockLevel() {
+        return ClusterBlockLevel.METADATA_READ;
+    }
+
     @Override
     protected ReplicationResponse newResponseInstance(StreamInput in) throws IOException {
         return new ReplicationResponse(in);

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.List;
@@ -48,15 +49,11 @@ public TransportRefreshAction(
             client,
             actionFilters,
             indexNameExpressionResolver,
-            TransportShardRefreshAction.TYPE
+            TransportShardRefreshAction.TYPE,
+            ThreadPool.Names.REFRESH
         );
     }
 
-    @Override
-    protected ReplicationResponse newShardResponse() {
-        return new ReplicationResponse();
-    }
-
     @Override
     protected BasicReplicationRequest newShardRequest(RefreshRequest request, ShardId shardId) {
         BasicReplicationRequest replicationRequest = new BasicReplicationRequest(shardId);

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -64,6 +65,11 @@ protected ReplicationResponse newResponseInstance(StreamInput in) throws IOExcep
         return new ReplicationResponse(in);
     }
 
+    @Override
+    protected ClusterBlockLevel globalBlockLevel() {
+        return ClusterBlockLevel.METADATA_READ;
+    }
+
     @Override
     protected void shardOperationOnPrimary(
         BasicReplicationRequest shardRequest,