Remove remaining TransportAction.execute calls from gateway code (#44095)

This commit removes most of the remaining direct execute calls, mostly
in gateway code. They are replaced with registering the actions as
internal actions.

relates #43881

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

@@ -71,6 +71,7 @@ import org.elasticsearch.action.admin.cluster.snapshots.get.TransportGetSnapshot
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotAction;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.TransportRestoreSnapshotAction;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusAction;
+import org.elasticsearch.action.admin.cluster.snapshots.status.TransportNodesSnapshotsStatus;
 import org.elasticsearch.action.admin.cluster.snapshots.status.TransportSnapshotsStatusAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
 import org.elasticsearch.action.admin.cluster.state.TransportClusterStateAction;
@@ -95,6 +96,7 @@ import org.elasticsearch.action.admin.indices.cache.clear.ClearIndicesCacheActio
 import org.elasticsearch.action.admin.indices.cache.clear.TransportClearIndicesCacheAction;
 import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
 import org.elasticsearch.action.admin.indices.close.TransportCloseIndexAction;
+import org.elasticsearch.action.admin.indices.close.TransportVerifyShardBeforeCloseAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
 import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
@@ -102,6 +104,7 @@ import org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction;
 import org.elasticsearch.action.admin.indices.flush.FlushAction;
 import org.elasticsearch.action.admin.indices.flush.SyncedFlushAction;
 import org.elasticsearch.action.admin.indices.flush.TransportFlushAction;
+import org.elasticsearch.action.admin.indices.flush.TransportShardFlushAction;
 import org.elasticsearch.action.admin.indices.flush.TransportSyncedFlushAction;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction;
 import org.elasticsearch.action.admin.indices.forcemerge.TransportForceMergeAction;
@@ -121,6 +124,7 @@ import org.elasticsearch.action.admin.indices.recovery.RecoveryAction;
 import org.elasticsearch.action.admin.indices.recovery.TransportRecoveryAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
 import org.elasticsearch.action.admin.indices.refresh.TransportRefreshAction;
+import org.elasticsearch.action.admin.indices.refresh.TransportShardRefreshAction;
 import org.elasticsearch.action.admin.indices.rollover.RolloverAction;
 import org.elasticsearch.action.admin.indices.rollover.TransportRolloverAction;
 import org.elasticsearch.action.admin.indices.segments.IndicesSegmentsAction;
@@ -206,11 +210,14 @@ import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.gateway.TransportNodesListGatewayMetaState;
+import org.elasticsearch.gateway.TransportNodesListGatewayStartedShards;
 import org.elasticsearch.index.seqno.GlobalCheckpointSyncAction;
 import org.elasticsearch.index.seqno.RetentionLeaseActions;
 import org.elasticsearch.index.seqno.RetentionLeaseBackgroundSyncAction;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncAction;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.indices.store.TransportNodesListShardStoreMetaData;
 import org.elasticsearch.persistent.CompletionPersistentTaskAction;
 import org.elasticsearch.persistent.RemovePersistentTaskAction;
 import org.elasticsearch.persistent.StartPersistentTaskAction;
@@ -532,6 +539,13 @@ public class ActionModule extends AbstractModule {
         actions.register(GlobalCheckpointSyncAction.TYPE, GlobalCheckpointSyncAction.class);
         actions.register(RetentionLeaseBackgroundSyncAction.TYPE, RetentionLeaseBackgroundSyncAction.class);
         actions.register(RetentionLeaseSyncAction.TYPE, RetentionLeaseSyncAction.class);
+        actions.register(TransportNodesSnapshotsStatus.TYPE, TransportNodesSnapshotsStatus.class);
+        actions.register(TransportNodesListGatewayMetaState.TYPE, TransportNodesListGatewayMetaState.class);
+        actions.register(TransportVerifyShardBeforeCloseAction.TYPE, TransportVerifyShardBeforeCloseAction.class);
+        actions.register(TransportNodesListGatewayStartedShards.TYPE, TransportNodesListGatewayStartedShards.class);
+        actions.register(TransportNodesListShardStoreMetaData.TYPE, TransportNodesListShardStoreMetaData.class);
+        actions.register(TransportShardFlushAction.TYPE, TransportShardFlushAction.class);
+        actions.register(TransportShardRefreshAction.TYPE, TransportShardRefreshAction.class);
 
         return unmodifiableMap(actions.getRegistry());
     }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java

@@ -20,7 +20,9 @@
 package org.elasticsearch.action.admin.cluster.snapshots.status;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.BaseNodeRequest;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
@@ -30,6 +32,7 @@ import org.elasticsearch.action.support.nodes.TransportNodesAction;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.index.shard.ShardId;
@@ -57,9 +60,16 @@ public class TransportNodesSnapshotsStatus extends TransportNodesAction<Transpor
                                                                         TransportNodesSnapshotsStatus.NodeSnapshotStatus> {
 
     public static final String ACTION_NAME = SnapshotsStatusAction.NAME + "[nodes]";
+    public static final ActionType<NodesSnapshotStatus> TYPE = new StreamableResponseActionType<>(ACTION_NAME) {
+        @Override
+        public NodesSnapshotStatus newResponse() {
+            return new TransportNodesSnapshotsStatus.NodesSnapshotStatus();
+        }
+    };
 
     private final SnapshotShardsService snapshotShardsService;
 
+    @Inject
     public TransportNodesSnapshotsStatus(ThreadPool threadPool, ClusterService clusterService,
                                          TransportService transportService, SnapshotShardsService snapshotShardsService,
                                          ActionFilters actionFilters) {
@@ -146,6 +156,10 @@ public class TransportNodesSnapshotsStatus extends TransportNodesAction<Transpor
 
     public static class NodesSnapshotStatus extends BaseNodesResponse<NodeSnapshotStatus> {
 
+        public NodesSnapshotStatus() {
+
+        }
+
         public NodesSnapshotStatus(ClusterName clusterName, List<NodeSnapshotStatus> nodes, List<FailedNodeException> failures) {
             super(clusterName, nodes, failures);
         }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java

@@ -24,6 +24,7 @@ import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.SnapshotsInProgress;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -61,17 +62,16 @@ public class TransportSnapshotsStatusAction extends TransportMasterNodeAction<Sn
 
     private final SnapshotsService snapshotsService;
 
-    private final TransportNodesSnapshotsStatus transportNodesSnapshotsStatus;
+    private final NodeClient client;
 
     @Inject
     public TransportSnapshotsStatusAction(TransportService transportService, ClusterService clusterService,
-                                          ThreadPool threadPool, SnapshotsService snapshotsService,
-                                          TransportNodesSnapshotsStatus transportNodesSnapshotsStatus,
+                                          ThreadPool threadPool, SnapshotsService snapshotsService, NodeClient client,
                                           ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
         super(SnapshotsStatusAction.NAME, transportService, clusterService, threadPool, actionFilters,
               SnapshotsStatusRequest::new, indexNameExpressionResolver);
         this.snapshotsService = snapshotsService;
-        this.transportNodesSnapshotsStatus = transportNodesSnapshotsStatus;
+        this.client = client;
     }
 
     @Override
@@ -119,7 +119,7 @@ public class TransportSnapshotsStatusAction extends TransportMasterNodeAction<Sn
             TransportNodesSnapshotsStatus.Request nodesRequest =
                 new TransportNodesSnapshotsStatus.Request(nodesIds.toArray(new String[nodesIds.size()]))
                     .snapshots(snapshots).timeout(request.masterNodeTimeout());
-            transportNodesSnapshotsStatus.execute(nodesRequest,
+            client.executeLocally(TransportNodesSnapshotsStatus.TYPE, nodesRequest,
                 ActionListener.map(
                     listener, nodeSnapshotStatuses ->
                         buildResponse(request, snapshotsService.currentSnapshots(request.repository(), Arrays.asList(request.snapshots())),

server/src/main/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseAction.java

@@ -22,6 +22,8 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationOperation;
@@ -52,6 +54,12 @@ public class TransportVerifyShardBeforeCloseAction extends TransportReplicationA
     TransportVerifyShardBeforeCloseAction.ShardRequest, TransportVerifyShardBeforeCloseAction.ShardRequest, ReplicationResponse> {
 
     public static final String NAME = CloseIndexAction.NAME + "[s]";
+    public static final ActionType<ReplicationResponse> TYPE = new StreamableResponseActionType<ReplicationResponse>(NAME) {
+        @Override
+        public ReplicationResponse newResponse() {
+            return new ReplicationResponse();
+        }
+    };
     protected Logger logger = LogManager.getLogger(getClass());
 
     @Inject

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

@@ -23,6 +23,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportBroadcastReplicationAction;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -38,11 +39,10 @@ public class TransportFlushAction
         extends TransportBroadcastReplicationAction<FlushRequest, FlushResponse, ShardFlushRequest, ReplicationResponse> {
 
     @Inject
-    public TransportFlushAction(ClusterService clusterService, TransportService transportService,
-                                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                                TransportShardFlushAction replicatedFlushAction) {
-        super(FlushAction.NAME, FlushRequest::new, clusterService, transportService, actionFilters, indexNameExpressionResolver,
-            replicatedFlushAction);
+    public TransportFlushAction(ClusterService clusterService, TransportService transportService, NodeClient client,
+                                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
+        super(FlushAction.NAME, FlushRequest::new, clusterService, transportService, client, actionFilters, indexNameExpressionResolver,
+            TransportShardFlushAction.TYPE);
     }
 
     @Override

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

@@ -20,6 +20,8 @@
 package org.elasticsearch.action.admin.indices.flush;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
@@ -40,6 +42,12 @@ public class TransportShardFlushAction
         extends TransportReplicationAction<ShardFlushRequest, ShardFlushRequest, ReplicationResponse> {
 
     public static final String NAME = FlushAction.NAME + "[s]";
+    public static final ActionType<ReplicationResponse> TYPE = new StreamableResponseActionType<ReplicationResponse>(NAME) {
+        @Override
+        public ReplicationResponse newResponse() {
+            return new ReplicationResponse();
+        }
+    };
 
     @Inject
     public TransportShardFlushAction(Settings settings, TransportService transportService, ClusterService clusterService,

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

@@ -25,6 +25,7 @@ import org.elasticsearch.action.support.DefaultShardOperationFailedException;
 import org.elasticsearch.action.support.replication.BasicReplicationRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportBroadcastReplicationAction;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -41,10 +42,9 @@ public class TransportRefreshAction
 
     @Inject
     public TransportRefreshAction(ClusterService clusterService, TransportService transportService, ActionFilters actionFilters,
-                                  IndexNameExpressionResolver indexNameExpressionResolver,
-                                  TransportShardRefreshAction shardRefreshAction) {
-        super(RefreshAction.NAME, RefreshRequest::new, clusterService, transportService, actionFilters,
-            indexNameExpressionResolver, shardRefreshAction);
+                                  IndexNameExpressionResolver indexNameExpressionResolver, NodeClient client) {
+        super(RefreshAction.NAME, RefreshRequest::new, clusterService, transportService, client, actionFilters,
+            indexNameExpressionResolver, TransportShardRefreshAction.TYPE);
     }
 
     @Override

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

@@ -20,6 +20,8 @@
 package org.elasticsearch.action.admin.indices.refresh;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.BasicReplicationRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -42,6 +44,12 @@ public class TransportShardRefreshAction
         extends TransportReplicationAction<BasicReplicationRequest, BasicReplicationRequest, ReplicationResponse> {
 
     public static final String NAME = RefreshAction.NAME + "[s]";
+    public static final ActionType<ReplicationResponse> TYPE = new StreamableResponseActionType<>(NAME) {
+        @Override
+        public ReplicationResponse newResponse() {
+            return new ReplicationResponse();
+        }
+    };
 
     @Inject
     public TransportShardRefreshAction(Settings settings, TransportService transportService, ClusterService clusterService,

server/src/main/java/org/elasticsearch/action/admin/indices/shards/TransportIndicesShardStoresAction.java

@@ -24,6 +24,8 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
+import org.elasticsearch.action.support.nodes.BaseNodesResponse;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -65,16 +67,15 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 public class TransportIndicesShardStoresAction
         extends TransportMasterNodeReadAction<IndicesShardStoresRequest, IndicesShardStoresResponse> {
 
-    private final TransportNodesListGatewayStartedShards listShardStoresInfo;
+    private final NodeClient client;
 
     @Inject
     public TransportIndicesShardStoresAction(TransportService transportService, ClusterService clusterService,
                                              ThreadPool threadPool, ActionFilters actionFilters,
-                                             IndexNameExpressionResolver indexNameExpressionResolver,
-                                             TransportNodesListGatewayStartedShards listShardStoresInfo) {
+                                             IndexNameExpressionResolver indexNameExpressionResolver, NodeClient client) {
         super(IndicesShardStoresAction.NAME, transportService, clusterService, threadPool, actionFilters,
             IndicesShardStoresRequest::new, indexNameExpressionResolver);
-        this.listShardStoresInfo = listShardStoresInfo;
+        this.client = client;
     }
 
     @Override
@@ -148,15 +149,23 @@ public class TransportIndicesShardStoresAction
                 listener.onResponse(new IndicesShardStoresResponse());
             } else {
                 for (ShardId shardId : shardIds) {
-                    InternalAsyncFetch fetch = new InternalAsyncFetch(logger, "shard_stores", shardId, listShardStoresInfo);
+                    InternalAsyncFetch fetch = new InternalAsyncFetch(logger, "shard_stores", shardId, this::listStartedShards);
                     fetch.fetchData(nodes, Collections.<String>emptySet());
                 }
             }
         }
 
+        private void listStartedShards(ShardId shardId, DiscoveryNode[] nodes,
+                                       ActionListener<BaseNodesResponse<NodeGatewayStartedShards>> listener) {
+            var request = new TransportNodesListGatewayStartedShards.Request(shardId, nodes);
+            client.executeLocally(TransportNodesListGatewayStartedShards.TYPE, request,
+                ActionListener.wrap(listener::onResponse, listener::onFailure));
+        }
+
         private class InternalAsyncFetch extends AsyncShardFetch<NodeGatewayStartedShards> {
 
-            InternalAsyncFetch(Logger logger, String type, ShardId shardId, TransportNodesListGatewayStartedShards action) {
+            InternalAsyncFetch(Logger logger, String type, ShardId shardId,
+                               Lister<? extends BaseNodesResponse<NodeGatewayStartedShards>, NodeGatewayStartedShards> action) {
                 super(logger, type, shardId, action);
             }
 

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

@@ -22,6 +22,7 @@ package org.elasticsearch.action.support.replication;
 import com.carrotsearch.hppc.cursors.IntObjectCursor;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
 import org.elasticsearch.action.support.HandledTransportAction;
@@ -29,6 +30,7 @@ import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.action.support.broadcast.BroadcastRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationFailedException;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -53,15 +55,17 @@ public abstract class TransportBroadcastReplicationAction<Request extends Broadc
         ShardRequest extends ReplicationRequest<ShardRequest>, ShardResponse extends ReplicationResponse>
         extends HandledTransportAction<Request, Response> {
 
-    private final TransportReplicationAction replicatedBroadcastShardAction;
+    private final ActionType<ShardResponse> replicatedBroadcastShardAction;
     private final ClusterService clusterService;
     private final IndexNameExpressionResolver indexNameExpressionResolver;
+    private final NodeClient client;
 
     public TransportBroadcastReplicationAction(String name, Writeable.Reader<Request> requestReader, ClusterService clusterService,
-                                               TransportService transportService,
+                                               TransportService transportService, NodeClient client,
                                                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                                               TransportReplicationAction replicatedBroadcastShardAction) {
+                                               ActionType<ShardResponse> replicatedBroadcastShardAction) {
         super(name, transportService, actionFilters, requestReader);
+        this.client = client;
         this.replicatedBroadcastShardAction = replicatedBroadcastShardAction;
         this.clusterService = clusterService;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
@@ -115,7 +119,7 @@ public abstract class TransportBroadcastReplicationAction<Request extends Broadc
     protected void shardExecute(Task task, Request request, ShardId shardId, ActionListener<ShardResponse> shardActionListener) {
         ShardRequest shardRequest = newShardRequest(request, shardId);
         shardRequest.setParentTask(clusterService.localNode().getId(), task.getId());
-        replicatedBroadcastShardAction.execute(shardRequest, shardActionListener);
+        client.executeLocally(replicatedBroadcastShardAction, shardRequest, shardActionListener);
     }
 
     /**

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java

@@ -35,6 +35,7 @@ import org.elasticsearch.action.admin.indices.close.TransportVerifyShardBeforeCl
 import org.elasticsearch.action.admin.indices.open.OpenIndexClusterStateUpdateRequest;
 import org.elasticsearch.action.support.ActiveShardsObserver;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
@@ -107,19 +108,18 @@ public class MetaDataIndexStateService {
     private final MetaDataIndexUpgradeService metaDataIndexUpgradeService;
     private final IndicesService indicesService;
     private final ThreadPool threadPool;
-    private final TransportVerifyShardBeforeCloseAction transportVerifyShardBeforeCloseAction;
+    private final NodeClient client;
     private final ActiveShardsObserver activeShardsObserver;
 
     @Inject
     public MetaDataIndexStateService(ClusterService clusterService, AllocationService allocationService,
                                      MetaDataIndexUpgradeService metaDataIndexUpgradeService,
-                                     IndicesService indicesService, ThreadPool threadPool,
-                                     TransportVerifyShardBeforeCloseAction transportVerifyShardBeforeCloseAction) {
+                                     IndicesService indicesService, ThreadPool threadPool, NodeClient client) {
         this.indicesService = indicesService;
         this.clusterService = clusterService;
         this.allocationService = allocationService;
         this.threadPool = threadPool;
-        this.transportVerifyShardBeforeCloseAction = transportVerifyShardBeforeCloseAction;
+        this.client = client;
         this.metaDataIndexUpgradeService = metaDataIndexUpgradeService;
         this.activeShardsObserver = new ActiveShardsObserver(clusterService, threadPool);
     }
@@ -401,7 +401,7 @@ public class MetaDataIndexStateService {
             if (request.ackTimeout() != null) {
                 shardRequest.timeout(request.ackTimeout());
             }
-            transportVerifyShardBeforeCloseAction.execute(shardRequest, new ActionListener<>() {
+            client.executeLocally(TransportVerifyShardBeforeCloseAction.TYPE, shardRequest, new ActionListener<>() {
                 @Override
                 public void onResponse(ReplicationResponse replicationResponse) {
                     final TransportVerifyShardBeforeCloseAction.ShardRequest shardRequest =
@@ -409,7 +409,7 @@ public class MetaDataIndexStateService {
                     if (request.ackTimeout() != null) {
                         shardRequest.timeout(request.ackTimeout());
                     }
-                    transportVerifyShardBeforeCloseAction.execute(shardRequest, listener);
+                    client.executeLocally(TransportVerifyShardBeforeCloseAction.TYPE, shardRequest, listener);
                 }
 
                 @Override

server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java

@@ -73,7 +73,6 @@ public abstract class AsyncShardFetch<T extends BaseNodeResponse> implements Rel
     private final AtomicLong round = new AtomicLong();
     private boolean closed;
 
-    @SuppressWarnings("unchecked")
     protected AsyncShardFetch(Logger logger, String type, ShardId shardId, Lister<? extends BaseNodesResponse<T>, T> action) {
         this.logger = logger;
         this.type = type;

server/src/main/java/org/elasticsearch/gateway/Gateway.java

@@ -24,6 +24,8 @@ import com.carrotsearch.hppc.cursors.ObjectCursor;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
@@ -39,17 +41,20 @@ public class Gateway {
 
     private final ClusterService clusterService;
 
-    private final TransportNodesListGatewayMetaState listGatewayMetaState;
+    private final NodeClient client;
 
-    public Gateway(final ClusterService clusterService, final TransportNodesListGatewayMetaState listGatewayMetaState) {
+    public Gateway(final ClusterService clusterService, final NodeClient client) {
         this.clusterService = clusterService;
-        this.listGatewayMetaState = listGatewayMetaState;
+        this.client = client;
     }
 
     public void performStateRecovery(final GatewayStateRecoveredListener listener) throws GatewayException {
         final String[] nodesIds = clusterService.state().nodes().getMasterNodes().keys().toArray(String.class);
         logger.trace("performing state recovery from {}", Arrays.toString(nodesIds));
-        final TransportNodesListGatewayMetaState.NodesGatewayMetaState nodesState = listGatewayMetaState.list(nodesIds, null).actionGet();
+        var request = new TransportNodesListGatewayMetaState.Request(nodesIds);
+        PlainActionFuture<TransportNodesListGatewayMetaState.NodesGatewayMetaState> future = PlainActionFuture.newFuture();
+        client.executeLocally(TransportNodesListGatewayMetaState.TYPE, request, future);
+        final TransportNodesListGatewayMetaState.NodesGatewayMetaState nodesState = future.actionGet();
 
         final int requiredAllocation = 1;
 

server/src/main/java/org/elasticsearch/gateway/GatewayAllocator.java

@@ -25,6 +25,8 @@ import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
 import org.elasticsearch.action.support.nodes.BaseNodesResponse;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -34,8 +36,10 @@ import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.gateway.TransportNodesListGatewayStartedShards.NodeGatewayStartedShards;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.store.TransportNodesListShardStoreMetaData;
+import org.elasticsearch.indices.store.TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData;
 
 import java.util.List;
 import java.util.concurrent.ConcurrentMap;
@@ -49,18 +53,16 @@ public class GatewayAllocator {
     private final PrimaryShardAllocator primaryShardAllocator;
     private final ReplicaShardAllocator replicaShardAllocator;
 
-    private final ConcurrentMap<ShardId, AsyncShardFetch<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards>>
+    private final ConcurrentMap<ShardId, AsyncShardFetch<NodeGatewayStartedShards>>
         asyncFetchStarted = ConcurrentCollections.newConcurrentMap();
-    private final ConcurrentMap<ShardId, AsyncShardFetch<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData>>
+    private final ConcurrentMap<ShardId, AsyncShardFetch<NodeStoreFilesMetaData>>
         asyncFetchStore = ConcurrentCollections.newConcurrentMap();
 
     @Inject
-    public GatewayAllocator(RerouteService rerouteService,
-                            TransportNodesListGatewayStartedShards startedAction,
-                            TransportNodesListShardStoreMetaData storeAction) {
+    public GatewayAllocator(RerouteService rerouteService, NodeClient client) {
         this.rerouteService = rerouteService;
-        this.primaryShardAllocator = new InternalPrimaryShardAllocator(startedAction);
-        this.replicaShardAllocator = new InternalReplicaShardAllocator(storeAction);
+        this.primaryShardAllocator = new InternalPrimaryShardAllocator(client);
+        this.replicaShardAllocator = new InternalReplicaShardAllocator(client);
     }
 
     public void cleanCaches() {
@@ -79,10 +81,10 @@ public class GatewayAllocator {
 
     public int getNumberOfInFlightFetch() {
         int count = 0;
-        for (AsyncShardFetch<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards> fetch : asyncFetchStarted.values()) {
+        for (AsyncShardFetch<NodeGatewayStartedShards> fetch : asyncFetchStarted.values()) {
             count += fetch.getNumberOfInFlightFetches();
         }
-        for (AsyncShardFetch<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> fetch : asyncFetchStore.values()) {
+        for (AsyncShardFetch<NodeStoreFilesMetaData> fetch : asyncFetchStore.values()) {
             count += fetch.getNumberOfInFlightFetches();
         }
         return count;
@@ -147,19 +149,18 @@ public class GatewayAllocator {
 
     class InternalPrimaryShardAllocator extends PrimaryShardAllocator {
 
-        private final TransportNodesListGatewayStartedShards startedAction;
+        private final NodeClient client;
 
-        InternalPrimaryShardAllocator(TransportNodesListGatewayStartedShards startedAction) {
-            this.startedAction = startedAction;
+        InternalPrimaryShardAllocator(NodeClient client) {
+            this.client = client;
         }
 
         @Override
-        protected AsyncShardFetch.FetchResult<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards>
-                                                                        fetchData(ShardRouting shard, RoutingAllocation allocation) {
-            AsyncShardFetch<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards> fetch =
+        protected AsyncShardFetch.FetchResult<NodeGatewayStartedShards> fetchData(ShardRouting shard, RoutingAllocation allocation) {
+            AsyncShardFetch<NodeGatewayStartedShards> fetch =
                 asyncFetchStarted.computeIfAbsent(shard.shardId(),
-                    shardId -> new InternalAsyncFetch<>(logger, "shard_started", shardId, startedAction));
-            AsyncShardFetch.FetchResult<TransportNodesListGatewayStartedShards.NodeGatewayStartedShards> shardState =
+                    shardId -> new InternalAsyncFetch<>(logger, "shard_started", shardId, this::listStartedShards));
+            AsyncShardFetch.FetchResult<NodeGatewayStartedShards> shardState =
                     fetch.fetchData(allocation.nodes(), allocation.getIgnoreNodes(shard.shardId()));
 
             if (shardState.hasData()) {
@@ -167,23 +168,30 @@ public class GatewayAllocator {
             }
             return shardState;
         }
+
+        private void listStartedShards(ShardId shardId, DiscoveryNode[] nodes,
+                                       ActionListener<BaseNodesResponse<NodeGatewayStartedShards>> listener) {
+            var request = new TransportNodesListGatewayStartedShards.Request(shardId, nodes);
+            client.executeLocally(TransportNodesListGatewayStartedShards.TYPE, request,
+                ActionListener.wrap(listener::onResponse, listener::onFailure));
+        }
     }
 
     class InternalReplicaShardAllocator extends ReplicaShardAllocator {
 
-        private final TransportNodesListShardStoreMetaData storeAction;
+        private final NodeClient client;
 
-        InternalReplicaShardAllocator(TransportNodesListShardStoreMetaData storeAction) {
-            this.storeAction = storeAction;
+        InternalReplicaShardAllocator(NodeClient client) {
+            this.client = client;
         }
 
         @Override
-        protected AsyncShardFetch.FetchResult<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData>
+        protected AsyncShardFetch.FetchResult<NodeStoreFilesMetaData>
                                                                         fetchData(ShardRouting shard, RoutingAllocation allocation) {
-            AsyncShardFetch<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> fetch =
+            AsyncShardFetch<NodeStoreFilesMetaData> fetch =
                 asyncFetchStore.computeIfAbsent(shard.shardId(),
-                    shardId -> new InternalAsyncFetch<>(logger, "shard_store", shard.shardId(), storeAction));
-            AsyncShardFetch.FetchResult<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> shardStores =
+                    shardId -> new InternalAsyncFetch<>(logger, "shard_store", shard.shardId(), this::listStoreFilesMetaData));
+            AsyncShardFetch.FetchResult<NodeStoreFilesMetaData> shardStores =
                     fetch.fetchData(allocation.nodes(), allocation.getIgnoreNodes(shard.shardId()));
             if (shardStores.hasData()) {
                 shardStores.processAllocation(allocation);
@@ -191,6 +199,13 @@ public class GatewayAllocator {
             return shardStores;
         }
 
+        private void listStoreFilesMetaData(ShardId shardId, DiscoveryNode[] nodes,
+                                            ActionListener<BaseNodesResponse<NodeStoreFilesMetaData>> listener) {
+            var request = new TransportNodesListShardStoreMetaData.Request(shardId, nodes);
+            client.executeLocally(TransportNodesListShardStoreMetaData.TYPE, request,
+                ActionListener.wrap(listener::onResponse, listener::onFailure));
+        }
+
         @Override
         protected boolean hasInitiatedFetching(ShardRouting shard) {
             return asyncFetchStore.get(shard.shardId()) != null;

server/src/main/java/org/elasticsearch/gateway/GatewayModule.java

@@ -28,8 +28,6 @@ public class GatewayModule extends AbstractModule {
     protected void configure() {
         bind(DanglingIndicesState.class).asEagerSingleton();
         bind(GatewayService.class).asEagerSingleton();
-        bind(TransportNodesListGatewayMetaState.class).asEagerSingleton();
-        bind(TransportNodesListGatewayStartedShards.class).asEagerSingleton();
         bind(LocalAllocateDangledIndices.class).asEagerSingleton();
     }
 }

server/src/main/java/org/elasticsearch/gateway/GatewayService.java

@@ -22,6 +22,7 @@ package org.elasticsearch.gateway;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
@@ -90,9 +91,7 @@ public class GatewayService extends AbstractLifecycleComponent implements Cluste
 
     @Inject
     public GatewayService(final Settings settings, final AllocationService allocationService, final ClusterService clusterService,
-                          final ThreadPool threadPool,
-                          final TransportNodesListGatewayMetaState listGatewayMetaState,
-                          final Discovery discovery) {
+                          final ThreadPool threadPool, final Discovery discovery, final NodeClient client) {
         this.allocationService = allocationService;
         this.clusterService = clusterService;
         this.threadPool = threadPool;
@@ -121,7 +120,7 @@ public class GatewayService extends AbstractLifecycleComponent implements Cluste
             recoveryRunnable = () ->
                     clusterService.submitStateUpdateTask("local-gateway-elected-state", new RecoverStateUpdateTask());
         } else {
-            final Gateway gateway = new Gateway(clusterService, listGatewayMetaState);
+            final Gateway gateway = new Gateway(clusterService, client);
             recoveryRunnable = () ->
                     gateway.performStateRecovery(new GatewayRecoveryListener());
         }

server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.gateway;
 
-import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.nodes.BaseNodeRequest;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
 import org.elasticsearch.action.support.nodes.BaseNodesRequest;
@@ -32,11 +32,9 @@ import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -50,6 +48,12 @@ public class TransportNodesListGatewayMetaState extends TransportNodesAction<Tra
                                                                              TransportNodesListGatewayMetaState.NodeGatewayMetaState> {
 
     public static final String ACTION_NAME = "internal:gateway/local/meta_state";
+    public static final ActionType<NodesGatewayMetaState> TYPE = new StreamableResponseActionType<>(ACTION_NAME) {
+        @Override
+        public NodesGatewayMetaState newResponse() {
+            return new NodesGatewayMetaState();
+        }
+    };
 
     private final GatewayMetaState metaState;
 
@@ -61,12 +65,6 @@ public class TransportNodesListGatewayMetaState extends TransportNodesAction<Tra
         this.metaState = metaState;
     }
 
-    public ActionFuture<NodesGatewayMetaState> list(String[] nodesIds, @Nullable TimeValue timeout) {
-        PlainActionFuture<NodesGatewayMetaState> future = PlainActionFuture.newFuture();
-        execute(new Request(nodesIds).timeout(timeout), future);
-        return future;
-    }
-
     @Override
     protected NodeRequest newNodeRequest(Request request) {
         return new NodeRequest();
@@ -99,6 +97,8 @@ public class TransportNodesListGatewayMetaState extends TransportNodesAction<Tra
 
     public static class NodesGatewayMetaState extends BaseNodesResponse<NodeGatewayMetaState> {
 
+        public NodesGatewayMetaState() {}
+
         public NodesGatewayMetaState(ClusterName clusterName, List<NodeGatewayMetaState> nodes, List<FailedNodeException> failures) {
             super(clusterName, nodes, failures);
         }

server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java

@@ -21,8 +21,9 @@ package org.elasticsearch.gateway;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.BaseNodeRequest;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
@@ -62,12 +63,15 @@ public class TransportNodesListGatewayStartedShards extends
     TransportNodesAction<TransportNodesListGatewayStartedShards.Request,
         TransportNodesListGatewayStartedShards.NodesGatewayStartedShards,
         TransportNodesListGatewayStartedShards.NodeRequest,
-        TransportNodesListGatewayStartedShards.NodeGatewayStartedShards>
-    implements
-    AsyncShardFetch.Lister<TransportNodesListGatewayStartedShards.NodesGatewayStartedShards,
         TransportNodesListGatewayStartedShards.NodeGatewayStartedShards> {
 
     public static final String ACTION_NAME = "internal:gateway/local/started_shards";
+    public static final ActionType<NodesGatewayStartedShards> TYPE = new StreamableResponseActionType<>(ACTION_NAME) {
+        @Override
+        public NodesGatewayStartedShards newResponse() {
+            return new NodesGatewayStartedShards();
+        }
+    };
     private final Settings settings;
     private final NodeEnvironment nodeEnv;
     private final IndicesService indicesService;
@@ -86,12 +90,6 @@ public class TransportNodesListGatewayStartedShards extends
         this.namedXContentRegistry = namedXContentRegistry;
     }
 
-    @Override
-    public void list(ShardId shardId, DiscoveryNode[] nodes,
-                     ActionListener<NodesGatewayStartedShards> listener) {
-        execute(new Request(shardId, nodes), listener);
-    }
-
     @Override
     protected NodeRequest newNodeRequest(Request request) {
         return new NodeRequest(request);
@@ -199,6 +197,8 @@ public class TransportNodesListGatewayStartedShards extends
 
     public static class NodesGatewayStartedShards extends BaseNodesResponse<NodeGatewayStartedShards> {
 
+        public NodesGatewayStartedShards() {}
+
         public NodesGatewayStartedShards(ClusterName clusterName, List<NodeGatewayStartedShards> nodes,
                                          List<FailedNodeException> failures) {
             super(clusterName, nodes, failures);

server/src/main/java/org/elasticsearch/indices/IndicesModule.java

@@ -60,7 +60,6 @@ import org.elasticsearch.indices.cluster.IndicesClusterStateService;
 import org.elasticsearch.indices.flush.SyncedFlushService;
 import org.elasticsearch.indices.mapper.MapperRegistry;
 import org.elasticsearch.indices.store.IndicesStore;
-import org.elasticsearch.indices.store.TransportNodesListShardStoreMetaData;
 import org.elasticsearch.plugins.MapperPlugin;
 
 import java.util.ArrayList;
@@ -242,7 +241,6 @@ public class IndicesModule extends AbstractModule {
         bind(IndicesStore.class).asEagerSingleton();
         bind(IndicesClusterStateService.class).asEagerSingleton();
         bind(SyncedFlushService.class).asEagerSingleton();
-        bind(TransportNodesListShardStoreMetaData.class).asEagerSingleton();
         bind(TransportResyncReplicationAction.class).asEagerSingleton();
         bind(PrimaryReplicaSyncer.class).asEagerSingleton();
     }

server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java

@@ -21,8 +21,9 @@ package org.elasticsearch.indices.store;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.StreamableResponseActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.BaseNodeRequest;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
@@ -41,7 +42,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.NodeEnvironment;
-import org.elasticsearch.gateway.AsyncShardFetch;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.IndexShard;
@@ -62,11 +62,15 @@ import java.util.concurrent.TimeUnit;
 public class TransportNodesListShardStoreMetaData extends TransportNodesAction<TransportNodesListShardStoreMetaData.Request,
     TransportNodesListShardStoreMetaData.NodesStoreFilesMetaData,
     TransportNodesListShardStoreMetaData.NodeRequest,
-    TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData>
-    implements AsyncShardFetch.Lister<TransportNodesListShardStoreMetaData.NodesStoreFilesMetaData,
     TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> {
 
     public static final String ACTION_NAME = "internal:cluster/nodes/indices/shard/store";
+    public static final ActionType<NodesStoreFilesMetaData> TYPE = new StreamableResponseActionType<>(ACTION_NAME) {
+        @Override
+        public NodesStoreFilesMetaData newResponse() {
+            return new NodesStoreFilesMetaData();
+        }
+    };
 
     private final Settings settings;
     private final IndicesService indicesService;
@@ -86,11 +90,6 @@ public class TransportNodesListShardStoreMetaData extends TransportNodesAction<T
         this.namedXContentRegistry = namedXContentRegistry;
     }
 
-    @Override
-    public void list(ShardId shardId, DiscoveryNode[] nodes, ActionListener<NodesStoreFilesMetaData> listener) {
-        execute(new Request(shardId, nodes), listener);
-    }
-
     @Override
     protected NodeRequest newNodeRequest(Request request) {
         return new NodeRequest(request);
@@ -268,6 +267,8 @@ public class TransportNodesListShardStoreMetaData extends TransportNodesAction<T
 
     public static class NodesStoreFilesMetaData extends BaseNodesResponse<NodeStoreFilesMetaData> {
 
+        public NodesStoreFilesMetaData() {}
+
         public NodesStoreFilesMetaData(ClusterName clusterName, List<NodeStoreFilesMetaData> nodes, List<FailedNodeException> failures) {
             super(clusterName, nodes, failures);
         }

server/src/main/java/org/elasticsearch/node/Node.java

@@ -28,7 +28,6 @@ import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionModule;
 import org.elasticsearch.action.ActionType;
-import org.elasticsearch.action.admin.cluster.snapshots.status.TransportNodesSnapshotsStatus;
 import org.elasticsearch.action.search.SearchExecutionStatsCollector;
 import org.elasticsearch.action.search.SearchPhaseController;
 import org.elasticsearch.action.search.SearchTransportService;
@@ -494,8 +493,6 @@ public class Node implements Closeable {
                 clusterModule.getIndexNameExpressionResolver(), repositoryService, threadPool);
             SnapshotShardsService snapshotShardsService = new SnapshotShardsService(settings, clusterService, snapshotsService, threadPool,
                 transportService, indicesService, actionModule.getActionFilters(), clusterModule.getIndexNameExpressionResolver());
-            TransportNodesSnapshotsStatus nodesSnapshotsStatus = new TransportNodesSnapshotsStatus(threadPool, clusterService,
-                transportService, snapshotShardsService, actionModule.getActionFilters());
             RestoreService restoreService = new RestoreService(clusterService, repositoryService, clusterModule.getAllocationService(),
                 metaDataCreateIndexService, metaDataIndexUpgradeService, clusterService.getClusterSettings());
 
@@ -581,7 +578,6 @@ public class Node implements Closeable {
                     b.bind(RepositoriesService.class).toInstance(repositoryService);
                     b.bind(SnapshotsService.class).toInstance(snapshotsService);
                     b.bind(SnapshotShardsService.class).toInstance(snapshotShardsService);
-                    b.bind(TransportNodesSnapshotsStatus.class).toInstance(nodesSnapshotsStatus);
                     b.bind(RestoreService.class).toInstance(restoreService);
                     b.bind(RerouteService.class).toInstance(rerouteService);
                 }

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

@@ -103,7 +103,7 @@ public class BroadcastReplicationTests extends ESTestCase {
         transportService.start();
         transportService.acceptIncomingRequests();
         broadcastReplicationAction = new TestBroadcastReplicationAction(clusterService, transportService,
-                new ActionFilters(new HashSet<>()), new IndexNameExpressionResolver(), null);
+                new ActionFilters(new HashSet<>()), new IndexNameExpressionResolver());
     }
 
     @Override
@@ -213,10 +213,9 @@ public class BroadcastReplicationTests extends ESTestCase {
             ConcurrentCollections.newConcurrentSet();
 
         TestBroadcastReplicationAction(ClusterService clusterService, TransportService transportService,
-                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                TransportReplicationAction<BasicReplicationRequest, BasicReplicationRequest, ReplicationResponse> action) {
+                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
             super("internal:test-broadcast-replication-action", DummyBroadcastRequest::new, clusterService, transportService,
-                    actionFilters, indexNameExpressionResolver, action);
+                    null, actionFilters, indexNameExpressionResolver, null);
         }
 
         @Override

server/src/test/java/org/elasticsearch/indices/cluster/ClusterStateChanges.java

@@ -24,6 +24,7 @@ import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteRequest;
 import org.elasticsearch.action.admin.cluster.reroute.TransportClusterRerouteAction;
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
@@ -41,9 +42,11 @@ import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.action.support.master.TransportMasterNodeActionUtils;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor.ClusterTasksResult;
@@ -187,11 +190,14 @@ public class ClusterStateChanges {
                 return indexMetaData;
             }
         };
+        NodeClient client = new NodeClient(Settings.EMPTY, threadPool);
+        Map<ActionType, TransportAction> actions = new HashMap<>();
+        actions.put(TransportVerifyShardBeforeCloseAction.TYPE, new TransportVerifyShardBeforeCloseAction(SETTINGS,
+            transportService, clusterService, indicesService, threadPool, null, actionFilters, indexNameExpressionResolver));
+        client.initialize(actions, null, null);
 
-        TransportVerifyShardBeforeCloseAction transportVerifyShardBeforeCloseAction = new TransportVerifyShardBeforeCloseAction(SETTINGS,
-            transportService, clusterService, indicesService, threadPool, null, actionFilters, indexNameExpressionResolver);
         MetaDataIndexStateService indexStateService = new MetaDataIndexStateService(clusterService, allocationService,
-            metaDataIndexUpgradeService, indicesService, threadPool, transportVerifyShardBeforeCloseAction);
+            metaDataIndexUpgradeService, indicesService, threadPool, client);
         MetaDataDeleteIndexService deleteIndexService = new MetaDataDeleteIndexService(SETTINGS, clusterService, allocationService);
         MetaDataUpdateSettingsService metaDataUpdateSettingsService = new MetaDataUpdateSettingsService(clusterService,
             allocationService, IndexScopedSettings.DEFAULT_SCOPED_SETTINGS, indicesService, threadPool);

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -1066,9 +1066,9 @@ public class SnapshotResiliencyTests extends ESTestCase {
             actions.put(IndicesShardStoresAction.INSTANCE,
                 new TransportIndicesShardStoresAction(
                     transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver,
-                    new TransportNodesListGatewayStartedShards(settings,
-                        threadPool, clusterService, transportService, actionFilters, nodeEnv, indicesService, namedXContentRegistry))
-            );
+                    client));
+            actions.put(TransportNodesListGatewayStartedShards.TYPE, new TransportNodesListGatewayStartedShards(settings,
+                threadPool, clusterService, transportService, actionFilters, nodeEnv, indicesService, namedXContentRegistry));
             actions.put(DeleteSnapshotAction.INSTANCE,
                 new TransportDeleteSnapshotAction(
                     transportService, clusterService, threadPool,

test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java

@@ -847,7 +847,7 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
                     getElectionStrategy());
                 masterService.setClusterStatePublisher(coordinator);
                 final GatewayService gatewayService = new GatewayService(settings, allocationService, clusterService,
-                    deterministicTaskQueue.getThreadPool(this::onNode), null, coordinator);
+                    deterministicTaskQueue.getThreadPool(this::onNode), coordinator, null);
 
                 logger.trace("starting up [{}]", localNode);
                 transportService.start();