Move unpromotable relocations to its own transport action

docs/changelog/127330.yaml

@@ -0,0 +1,5 @@
+pr: 127330
+summary: Move unpromotable relocations to its own transport action
+area: Recovery
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java

@@ -301,18 +301,21 @@ private void doRecovery(final long recoveryId, final StartRecoveryRequest preExi
         if (indexShard.routingEntry().isPromotableToPrimary() == false) {
             assert preExistingRequest == null;
             assert indexShard.indexSettings().getIndexMetadata().isSearchableSnapshot() == false;
-            ActionListener.run(cleanupOnly.map(v -> {
-                logger.trace("{} preparing unpromotable shard for recovery", recoveryTarget.shardId());
-                indexShard.prepareForIndexRecovery();
-                // Skip unnecessary intermediate stages
-                recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
-                recoveryState.setStage(RecoveryState.Stage.TRANSLOG);
-                indexShard.openEngineAndSkipTranslogRecovery();
-                recoveryState.getIndex().setFileDetailsComplete();
-                recoveryState.setStage(RecoveryState.Stage.FINALIZE);
-                onGoingRecoveries.markRecoveryAsDone(recoveryId);
-                return null;
-            }), indexShard::preRecovery);
+            client.execute(
+                StatelessUnpromotableRelocationAction.TYPE,
+                new StatelessUnpromotableRelocationAction.Request(
+                    recoveryId,
+                    indexShard.shardId(),
+                    indexShard.routingEntry().allocationId().getId(),
+                    recoveryTarget.clusterStateVersion()
+                ),
+                cleanupOnly.delegateFailure((l, unused) -> {
+                    ActionListener.completeWith(l, () -> {
+                        onGoingRecoveries.markRecoveryAsDone(recoveryId);
+                        return null;
+                    });
+                })
+            );
             return;
         }
 

server/src/main/java/org/elasticsearch/indices/recovery/RecoveriesCollection.java

@@ -167,7 +167,6 @@ public RecoveryRef getRecoverySafe(long id, ShardId shardId) {
             throw new IndexShardClosedException(shardId);
         }
         assert recoveryRef.target().shardId().equals(shardId);
-        assert recoveryRef.target().indexShard().routingEntry().isPromotableToPrimary();
         return recoveryRef;
     }
 

server/src/main/java/org/elasticsearch/indices/recovery/StatelessUnpromotableRelocationAction.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.indices.recovery;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.shard.ShardId;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class StatelessUnpromotableRelocationAction {
+
+    public static final ActionType<ActionResponse.Empty> TYPE = new ActionType<>(
+        "internal:index/shard/recovery/stateless_unpromotable_relocation"
+    );
+
+    public static class Request extends ActionRequest {
+        private final long recoveryId;
+        private final ShardId shardId;
+        private final String targetAllocationId;
+        private final long clusterStateVersion;
+
+        public Request(long recoveryId, ShardId shardId, String targetAllocationId, long clusterStateVersion) {
+            this.recoveryId = recoveryId;
+            this.shardId = shardId;
+            this.targetAllocationId = targetAllocationId;
+            this.clusterStateVersion = clusterStateVersion;
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            recoveryId = in.readVLong();
+            shardId = new ShardId(in);
+            targetAllocationId = in.readString();
+            clusterStateVersion = in.readVLong();
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeVLong(recoveryId);
+            shardId.writeTo(out);
+            out.writeString(targetAllocationId);
+            out.writeVLong(clusterStateVersion);
+        }
+
+        public long getRecoveryId() {
+            return recoveryId;
+        }
+
+        public ShardId getShardId() {
+            return shardId;
+        }
+
+        public long getClusterStateVersion() {
+            return clusterStateVersion;
+        }
+
+        public String getTargetAllocationId() {
+            return targetAllocationId;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return recoveryId == request.recoveryId
+                && clusterStateVersion == request.clusterStateVersion
+                && Objects.equals(shardId, request.shardId)
+                && Objects.equals(targetAllocationId, request.targetAllocationId);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(recoveryId, shardId, targetAllocationId, clusterStateVersion);
+        }
+    }
+}

server/src/test/java/org/elasticsearch/indices/recovery/StatelessUnpromotableRelocationActionTests.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.indices.recovery;
+
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.shard.ShardIdTests;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
+
+import java.io.IOException;
+
+public class StatelessUnpromotableRelocationActionTests extends AbstractWireSerializingTestCase<
+    StatelessUnpromotableRelocationAction.Request> {
+    @Override
+    protected Writeable.Reader<StatelessUnpromotableRelocationAction.Request> instanceReader() {
+        return StatelessUnpromotableRelocationAction.Request::new;
+    }
+
+    @Override
+    protected StatelessUnpromotableRelocationAction.Request createTestInstance() {
+        return new StatelessUnpromotableRelocationAction.Request(
+            randomNonNegativeLong(),
+            new ShardId(randomIdentifier(), UUIDs.randomBase64UUID(), randomIntBetween(0, 99)),
+            randomUUID(),
+            randomNonNegativeLong()
+        );
+    }
+
+    @Override
+    protected StatelessUnpromotableRelocationAction.Request mutateInstance(StatelessUnpromotableRelocationAction.Request instance)
+        throws IOException {
+        return switch (between(0, 3)) {
+            case 0 -> new StatelessUnpromotableRelocationAction.Request(
+                randomValueOtherThan(instance.getRecoveryId(), ESTestCase::randomNonNegativeLong),
+                instance.getShardId(),
+                instance.getTargetAllocationId(),
+                instance.getClusterStateVersion()
+            );
+            case 1 -> new StatelessUnpromotableRelocationAction.Request(
+                instance.getRecoveryId(),
+                ShardIdTests.mutate(instance.getShardId()),
+                instance.getTargetAllocationId(),
+                instance.getClusterStateVersion()
+            );
+            case 2 -> new StatelessUnpromotableRelocationAction.Request(
+                instance.getRecoveryId(),
+                instance.getShardId(),
+                randomValueOtherThan(instance.getTargetAllocationId(), ESTestCase::randomUUID),
+                instance.getClusterStateVersion()
+            );
+            case 3 -> new StatelessUnpromotableRelocationAction.Request(
+                instance.getRecoveryId(),
+                instance.getShardId(),
+                instance.getTargetAllocationId(),
+                randomValueOtherThan(instance.getClusterStateVersion(), ESTestCase::randomNonNegativeLong)
+            );
+            default -> throw new AssertionError("Illegal randomisation branch");
+        };
+    }
+}