Add unpromotable transport action implementation for tests

Author: fcofdez

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/ShardRoutingRoleIT.java

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteUtils;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsGroup;
@@ -21,6 +22,8 @@
 import org.elasticsearch.action.search.OpenPointInTimeRequest;
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
 import org.elasticsearch.action.search.TransportOpenPointInTimeAction;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
@@ -38,6 +41,7 @@
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexService;
@@ -48,12 +52,18 @@
 import org.elasticsearch.index.engine.NoOpEngine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.indices.recovery.PeerRecoveryTargetService;
+import org.elasticsearch.indices.recovery.RecoveryState;
+import org.elasticsearch.indices.recovery.StatelessUnpromotableRelocationAction;
+import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.ClusterPlugin;
 import org.elasticsearch.plugins.EnginePlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.search.builder.PointInTimeBuilder;
 import org.elasticsearch.snapshots.SnapshotState;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.XContentTestUtils;
 import org.elasticsearch.test.transport.MockTransportService;
@@ -91,7 +101,7 @@ public class ShardRoutingRoleIT extends ESIntegTestCase {
 
     private static final Logger logger = LogManager.getLogger(ShardRoutingRoleIT.class);
 
-    public static class TestPlugin extends Plugin implements ClusterPlugin, EnginePlugin {
+    public static class TestPlugin extends Plugin implements ClusterPlugin, EnginePlugin, ActionPlugin {
 
         volatile int numIndexingCopies = 1;
         static final String NODE_ATTR_UNPROMOTABLE_ONLY = "unpromotableonly";
@@ -112,6 +122,61 @@ public ShardRouting.Role newEmptyRole(int copyIndex) {
             };
         }
 
+        // This is implemented in stateless, but for the tests we need to provide a simple implementation
+        public static class TransportTestUnpromotableRelocationAction extends TransportAction<
+            StatelessUnpromotableRelocationAction.Request,
+            ActionResponse.Empty> {
+
+            private final IndicesService indicesService;
+            private final PeerRecoveryTargetService peerRecoveryTargetService;
+
+            @Inject
+            public TransportTestUnpromotableRelocationAction(
+                ActionFilters actionFilters,
+                IndicesService indicesService,
+                PeerRecoveryTargetService peerRecoveryTargetService,
+                TransportService transportService
+            ) {
+                super(
+                    StatelessUnpromotableRelocationAction.TYPE.name(),
+                    actionFilters,
+                    transportService.getTaskManager(),
+                    EsExecutors.DIRECT_EXECUTOR_SERVICE
+                );
+                this.indicesService = indicesService;
+                this.peerRecoveryTargetService = peerRecoveryTargetService;
+            }
+
+            @Override
+            protected void doExecute(
+                Task task,
+                StatelessUnpromotableRelocationAction.Request request,
+                ActionListener<ActionResponse.Empty> listener
+            ) {
+                try (var recoveryRef = peerRecoveryTargetService.getRecoveryRef(request.getRecoveryId(), request.getShardId())) {
+                    final var indexService = indicesService.indexServiceSafe(request.getShardId().getIndex());
+                    final var indexShard = indexService.getShard(request.getShardId().id());
+                    final var recoveryTarget = recoveryRef.target();
+                    final var recoveryState = recoveryTarget.state();
+
+                    ActionListener.completeWith(listener, () -> {
+                        indexShard.prepareForIndexRecovery();
+                        recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
+                        recoveryState.setStage(RecoveryState.Stage.TRANSLOG);
+                        indexShard.openEngineAndSkipTranslogRecovery();
+                        recoveryState.getIndex().setFileDetailsComplete();
+                        recoveryState.setStage(RecoveryState.Stage.FINALIZE);
+                        return ActionResponse.Empty.INSTANCE;
+                    });
+                }
+            }
+        }
+
+        @Override
+        public Collection<ActionHandler> getActions() {
+            return List.of(new ActionHandler(StatelessUnpromotableRelocationAction.TYPE, TransportTestUnpromotableRelocationAction.class));
+        }
+
         @Override
         public Collection<AllocationDecider> createAllocationDeciders(Settings settings, ClusterSettings clusterSettings) {
             return List.of(new AllocationDecider() {