Add new recovery source for reshard split target shards

Author: lkts

server/src/main/java/org/elasticsearch/cluster/routing/ExpectedShardSizeEstimator.java

@@ -58,6 +58,7 @@ public static boolean shouldReserveSpaceForInitializingShard(ShardRouting shard,
             // shrink/split/clone operation is going to clone existing locally placed shards using file system hard links
             // so no additional space is going to be used until future merges
             case LOCAL_SHARDS -> false;
+            case RESHARD_SPLIT_TARGET -> false;
         };
     }
 

server/src/main/java/org/elasticsearch/cluster/routing/RecoverySource.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.xcontent.ToXContent;
@@ -31,6 +32,7 @@
  * - {@link PeerRecoverySource} recovery from a primary on another node
  * - {@link SnapshotRecoverySource} recovery from a snapshot
  * - {@link LocalShardsRecoverySource} recovery from other shards of another index on the same node
+ * - {@link SplitTargetRecoverySource} recovery of a shard that is created as a result of a resharding split
  */
 public abstract class RecoverySource implements Writeable, ToXContentObject {
 
@@ -50,13 +52,16 @@ public void addAdditionalFields(XContentBuilder builder, ToXContent.Params param
     }
 
     public static RecoverySource readFrom(StreamInput in) throws IOException {
+        // TODO is transport version check needed?
+
         Type type = Type.values()[in.readByte()];
         return switch (type) {
             case EMPTY_STORE -> EmptyStoreRecoverySource.INSTANCE;
             case EXISTING_STORE -> ExistingStoreRecoverySource.read(in);
             case PEER -> PeerRecoverySource.INSTANCE;
             case SNAPSHOT -> new SnapshotRecoverySource(in);
             case LOCAL_SHARDS -> LocalShardsRecoverySource.INSTANCE;
+            case RESHARD_SPLIT_TARGET -> new SplitTargetRecoverySource(in);
         };
     }
 
@@ -78,7 +83,8 @@ public enum Type {
         EXISTING_STORE,
         PEER,
         SNAPSHOT,
-        LOCAL_SHARDS
+        LOCAL_SHARDS,
+        RESHARD_SPLIT_TARGET
     }
 
     public abstract Type getType();
@@ -319,4 +325,39 @@ public boolean expectEmptyRetentionLeases() {
             return false;
         }
     }
+
+    /**
+     * Recovery of a shard that is created as a result of a resharding split.
+     * Not to be confused with _split API.
+     */
+    public static class SplitTargetRecoverySource extends RecoverySource {
+        private final ShardId sourceShardId;
+
+        public SplitTargetRecoverySource(ShardId sourceShardId) {
+            this.sourceShardId = sourceShardId;
+        }
+
+        SplitTargetRecoverySource(StreamInput in) throws IOException {
+            sourceShardId = new ShardId(in);
+        }
+
+        @Override
+        public Type getType() {
+            return Type.RESHARD_SPLIT_TARGET;
+        }
+
+        public ShardId getSourceShardId() {
+            return sourceShardId;
+        }
+
+        @Override
+        protected void writeAdditionalFields(StreamOutput out) throws IOException {
+            sourceShardId.writeTo(out);
+        }
+
+        @Override
+        public void addAdditionalFields(XContentBuilder builder, Params params) throws IOException {
+            sourceShardId.toXContent(builder, params);
+        }
+    }
 }

server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
+import org.elasticsearch.cluster.routing.RecoverySource;
 import org.elasticsearch.cluster.routing.RecoverySource.Type;
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.RoutingTable;
@@ -701,6 +702,14 @@ private void createShard(ShardRouting shardRouting, ClusterState state) {
                     logger.trace("ignoring initializing shard {} - no source node can be found.", shardId);
                     return;
                 }
+            } else if (shardRouting.recoverySource().getType() == Type.RESHARD_SPLIT_TARGET) {
+                ShardId sourceShardId = ((RecoverySource.SplitTargetRecoverySource) shardRouting.recoverySource()).getSourceShardId();
+                sourceNode = findSourceNodeForSplitTargetRecovery(
+                    state.routingTable(project.id()),
+                    state.nodes(),
+                    sourceShardId
+
+                );
             } else {
                 sourceNode = null;
             }
@@ -988,6 +997,29 @@ private static DiscoveryNode findSourceNodeForPeerRecovery(RoutingTable routingT
         return sourceNode;
     }
 
+    private static DiscoveryNode findSourceNodeForSplitTargetRecovery(
+        RoutingTable routingTable,
+        DiscoveryNodes nodes,
+        ShardId sourceShardId
+    ) {
+        ShardRouting sourceShardRouting = routingTable.shardRoutingTable(sourceShardId).primaryShard();
+
+        if (sourceShardRouting.active() == false) {
+            logger.trace("can't find reshard split source node because source shard {} is not active.", sourceShardRouting);
+            return null;
+        }
+
+        DiscoveryNode sourceNode = nodes.get(sourceShardRouting.currentNodeId());
+        if (sourceNode == null) {
+            logger.trace(
+                "can't find reshard split source node because source shard {} is assigned to an unknown node.",
+                sourceShardRouting
+            );
+            return null;
+        }
+        return sourceNode;
+    }
+
     private record PendingShardCreation(String clusterStateUUID, long startTimeMillis) {}
 
     private class RecoveryListener implements PeerRecoveryTargetService.RecoveryListener {

server/src/test/java/org/elasticsearch/cluster/routing/RecoverySourceTests.java

@@ -34,10 +34,11 @@ public void testRecoverySourceTypeOrder() {
         assertEquals(RecoverySource.Type.PEER.ordinal(), 2);
         assertEquals(RecoverySource.Type.SNAPSHOT.ordinal(), 3);
         assertEquals(RecoverySource.Type.LOCAL_SHARDS.ordinal(), 4);
+        assertEquals(RecoverySource.Type.RESHARD_SPLIT_TARGET.ordinal(), 5);
         // check exhaustiveness
         for (RecoverySource.Type type : RecoverySource.Type.values()) {
             assertThat(type.ordinal(), greaterThanOrEqualTo(0));
-            assertThat(type.ordinal(), lessThanOrEqualTo(4));
+            assertThat(type.ordinal(), lessThanOrEqualTo(5));
         }
     }
 }

test/framework/src/main/java/org/elasticsearch/cluster/routing/TestShardRouting.java

@@ -275,7 +275,8 @@ public static RecoverySource buildRecoverySource() {
                 new Snapshot("repo", new SnapshotId(randomIdentifier(), randomUUID())),
                 IndexVersion.current(),
                 new IndexId("some_index", randomUUID())
-            )
+            ),
+            new RecoverySource.SplitTargetRecoverySource(new ShardId("some_index", randomUUID(), randomIntBetween(0, 1000)))
         );
     }
 }