Read reshardShardCount

Author: ankikuma

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

@@ -159,6 +159,14 @@ public ShardId shardId() {
         return shardId;
     }
 
+    /**
+     * @return The effective shard count as seen by the coordinator when creating this request.
+     * can be 0 if this has not yet been resolved.
+     */
+    public int reshardSplitShardCount() {
+        return reshardSplitShardCount;
+    }
+
     /**
      * Sets the number of shard copies that must be active before proceeding with the replication
      * operation. Defaults to {@link ActiveShardCount#DEFAULT}, which requires one shard copy

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexReshardingState;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -459,14 +460,22 @@ void runWithPrimaryShardReference(final PrimaryShardReference primaryShardRefere
             try {
                 final ClusterState clusterState = clusterService.state();
                 final Index index = primaryShardReference.routingEntry().index();
-                final ProjectId projectId = clusterState.metadata().projectFor(index).id();
+                final ProjectMetadata project = clusterState.metadata().projectFor(index);
+                final ProjectId projectId = project.id();
+                final IndexMetadata indexMetadata = project.index(index);
 
                 final ClusterBlockException blockException = blockExceptions(clusterState, projectId, index.getName());
                 if (blockException != null) {
                     logger.trace("cluster is blocked, action failed on primary", blockException);
                     throw blockException;
                 }
 
+                int reshardSplitShardCount = primaryRequest.getRequest().reshardSplitShardCount();
+                assert (reshardSplitShardCount == 0
+                    || reshardSplitShardCount == indexMetadata.getReshardSplitShardCount(
+                        primaryRequest.getRequest().shardId().getId(),
+                        IndexReshardingState.Split.TargetShardState.HANDOFF
+                    ));
                 if (primaryShardReference.isRelocated()) {
                     primaryShardReference.close(); // release shard operation lock as soon as possible
                     setPhase(replicationTask, "primary_delegation");