Changes

Author: Tim-Brooks

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

@@ -893,8 +893,18 @@ public IndexMetadata withInSyncAllocationIds(int shardId, Set<String> inSyncSet)
      * @return updated instance with incremented primary term
      */
     public IndexMetadata withIncrementedPrimaryTerm(int shardId) {
+        return withSetPrimaryTerm(shardId, this.primaryTerms[shardId] + 1);
+    }
+
+    /**
+     * Creates a copy of this instance that has the primary term for the given shard id set to the value provided.
+     * @param shardId shard id to set primary term for
+     * @param primaryTerm primary term to set
+     * @return updated instance with set primary term
+     */
+    public IndexMetadata withSetPrimaryTerm(int shardId, long primaryTerm) {
         final long[] incremented = this.primaryTerms.clone();
-        incremented[shardId]++;
+        incremented[shardId] = primaryTerm;
         return new IndexMetadata(
             this.index,
             this.version,

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

@@ -12,13 +12,15 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentFragment;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Objects;
 
 /**
@@ -205,6 +207,18 @@ public static IndexReshardingMetadata newSplitByMultiple(int shardCount, int mul
         return new IndexReshardingMetadata(IndexReshardingState.Split.newSplitByMultiple(shardCount, multiple));
     }
 
+    public IndexReshardingMetadata transitionSplitTargetToHandoff(ShardId shardId) {
+        assert state instanceof IndexReshardingState.Split;
+        IndexReshardingState.Split splitState = (IndexReshardingState.Split) state;
+        IndexReshardingState.Split.TargetShardState[] newTargets = Arrays.copyOf(
+            splitState.targetShards(),
+            splitState.targetShards().length
+        );
+        int i = shardId.getId() - state.shardCountBefore();
+        newTargets[i] = IndexReshardingState.Split.TargetShardState.HANDOFF;
+        return new IndexReshardingMetadata(new IndexReshardingState.Split(splitState.sourceShards(), newTargets));
+    }
+
     /**
      * @return the split state of this metadata block, or throw IllegalArgumentException if this metadata doesn't represent a split
      */

server/src/main/java/org/elasticsearch/cluster/routing/allocation/IndexMetadataUpdater.java

@@ -12,9 +12,11 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexReshardingMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.routing.GlobalRoutingTable;
+import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RecoverySource;
 import org.elasticsearch.cluster.routing.RoutingChangesObserver;
@@ -127,15 +129,19 @@ public Metadata applyChanges(Metadata oldMetadata, GlobalRoutingTable newRouting
                 for (Map.Entry<ShardId, Updates> shardEntry : indexChanges) {
                     ShardId shardId = shardEntry.getKey();
                     Updates updates = shardEntry.getValue();
-                    updatedIndexMetadata = updateInSyncAllocations(
-                        newRoutingTable.routingTable(projectMetadata.id()),
-                        oldIndexMetadata,
-                        updatedIndexMetadata,
-                        shardId,
-                        updates
-                    );
+                    RoutingTable routingTable = newRoutingTable.routingTable(projectMetadata.id());
+                    updatedIndexMetadata = updateInSyncAllocations(routingTable, oldIndexMetadata, updatedIndexMetadata, shardId, updates);
+                    IndexRoutingTable indexRoutingTable = routingTable.index(shardEntry.getKey().getIndex());
+                    RecoverySource recoverySource = indexRoutingTable.shard(shardEntry.getKey().id()).primaryShard().recoverySource();
+                    IndexReshardingMetadata reshardingMetadata = updatedIndexMetadata.getReshardingMetadata();
+                    boolean split = recoverySource != null && reshardingMetadata != null;
                     updatedIndexMetadata = updates.increaseTerm
-                        ? updatedIndexMetadata.withIncrementedPrimaryTerm(shardId.id())
+                        ? split
+                            ? updatedIndexMetadata.withSetPrimaryTerm(
+                                shardId.id(),
+                                splitPrimaryTerm(updatedIndexMetadata, reshardingMetadata, shardId)
+                            )
+                            : updatedIndexMetadata.withIncrementedPrimaryTerm(shardId.id())
                         : updatedIndexMetadata;
                 }
                 if (updatedIndexMetadata != oldIndexMetadata) {
@@ -147,6 +153,15 @@ public Metadata applyChanges(Metadata oldMetadata, GlobalRoutingTable newRouting
         return updatedMetadata.build();
     }
 
+    private static long splitPrimaryTerm(IndexMetadata updatedIndexMetadata, IndexReshardingMetadata reshardingMetadata, ShardId shardId) {
+        // We take the max of the source and target primary terms. This guarantees that the target primary term stays
+        // greater than or equal to the source.
+        return Math.max(
+            updatedIndexMetadata.primaryTerm(shardId.getId() % reshardingMetadata.shardCountBefore()),
+            updatedIndexMetadata.primaryTerm(shardId.id()) + 1
+        );
+    }
+
     /**
      * Updates in-sync allocations with routing changes that were made to the routing table.
      */