fix

Author: lkts

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

@@ -47,6 +47,21 @@
  * Note that in this case no shard-bulk-request is sent to shards 5, 6, 7 and the requests that were meant for these target shards
  * are bundled together with and sent to their source shards.
  *
+ * Example 3:
+ * Suppose we are resharding an index from 4 -> 8 shards. While handling a search request, the coordinator observes
+ * that target shard 5 is in SPLIT state but target shards 4, 6, 7 are in CLONE/HANDOFF state.
+ * The coordinator will send shard search requests to all source shards (0, 1, 2, 3) and to all target shards
+ * that are at least in SPLIT state (5).
+ * Shard search request sent to source shards 0, 2, 3 has the "reshardSplitShardCountSummary" of 4
+ * since corresponding target shards (4, 6, 7) have not advanced to SPLIT state.
+ * Shard search request sent to source shard 1 has the "reshardSplitShardCountSummary" of 8
+ * since the corresponding target shard 5 is in SPLIT state.
+ * When a shard search request is executed on the source shard 1, "reshardSplitShardCountSummary" value
+ * is checked and documents that will be returned by target shard 5 are excluded
+ * (they are still present in the source shard because the resharding process is not complete).
+ * All other source shard search requests (0, 2, 3) return all available documents since corresponding target shards
+ * are not yet available to do that.
+ *
  * A value of 0 indicates an INVALID reshardSplitShardCountSummary. Hence, a request with INVALID reshardSplitShardCountSummary
  * will be treated as a Summary mismatch on the source shard node.
  */

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

@@ -524,7 +524,7 @@ public void testOperationRoutingWithResharding() throws IOException {
         assertEquals(shardCount, initialSearchShards.size());
         for (int i = 0; i < shardCount; i++) {
             assertEquals(i, initialSearchShards.get(i).iterator().shardId().id());
-            assertEquals(shardCount, initialSearchShards.get(i).reshardSplitShardCountSummary());
+            assertEquals(SplitShardCountSummary.fromInt(shardCount), initialSearchShards.get(i).reshardSplitShardCountSummary());
         }
 
         // We are testing a case when there is routing configuration but not for the index in question.
@@ -534,7 +534,7 @@ public void testOperationRoutingWithResharding() throws IOException {
         assertEquals(shardCount, initialSearchShardsWithRouting.size());
         for (int i = 0; i < shardCount; i++) {
             assertEquals(i, initialSearchShardsWithRouting.get(i).iterator().shardId().id());
-            assertEquals(shardCount, initialSearchShardsWithRouting.get(i).reshardSplitShardCountSummary());
+            assertEquals(SplitShardCountSummary.fromInt(shardCount), initialSearchShardsWithRouting.get(i).reshardSplitShardCountSummary());
         }
 
         var initialWriteableShards = clusterService.operationRouting()
@@ -571,15 +571,21 @@ public void testOperationRoutingWithResharding() throws IOException {
         assertEquals(shardCount, searchShardsWithOneShardHandoff.size());
         for (int i = 0; i < shardCount; i++) {
             assertEquals(i, searchShardsWithOneShardHandoff.get(i).iterator().shardId().id());
-            assertEquals(shardCount, searchShardsWithOneShardHandoff.get(i).reshardSplitShardCountSummary());
+            assertEquals(
+                SplitShardCountSummary.fromInt(shardCount),
+                searchShardsWithOneShardHandoff.get(i).reshardSplitShardCountSummary()
+            );
         }
 
         var searchShardsWithOneShardHandoffAndRouting = clusterService.operationRouting()
             .searchShards(clusterService.state().projectState(projectId), new String[] { indexName }, Map.of("other", Set.of("1")), null);
         assertEquals(shardCount, searchShardsWithOneShardHandoffAndRouting.size());
         for (int i = 0; i < shardCount; i++) {
             assertEquals(i, searchShardsWithOneShardHandoffAndRouting.get(i).iterator().shardId().id());
-            assertEquals(shardCount, searchShardsWithOneShardHandoffAndRouting.get(i).reshardSplitShardCountSummary());
+            assertEquals(
+                SplitShardCountSummary.fromInt(shardCount),
+                searchShardsWithOneShardHandoffAndRouting.get(i).reshardSplitShardCountSummary()
+            );
         }
 
         var writeableShardsWithOneShardHandoff = clusterService.operationRouting()
@@ -622,12 +628,21 @@ public void testOperationRoutingWithResharding() throws IOException {
         }
         assertEquals(shardChangingSplitTargetState, searchShardsWithOneShardSplit.get(indexOfShardWithNewState).iterator().shardId().id());
         // Since the target shard is in SPLIT state, reshardSplitShardCountSummary is updated for it and the corresponding source shard.
-        assertEquals(newShardCount, searchShardsWithOneShardSplit.get(indexOfShardWithNewState).reshardSplitShardCountSummary());
+        assertEquals(
+            SplitShardCountSummary.fromInt(newShardCount),
+            searchShardsWithOneShardSplit.get(indexOfShardWithNewState).reshardSplitShardCountSummary()
+        );
         for (int i = 0; i < shardCount; i++) {
             if (i == sourceShard) {
-                assertEquals(newShardCount, searchShardsWithOneShardSplit.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(newShardCount),
+                    searchShardsWithOneShardSplit.get(i).reshardSplitShardCountSummary()
+                );
             } else {
-                assertEquals(shardCount, searchShardsWithOneShardSplit.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(shardCount),
+                    searchShardsWithOneShardSplit.get(i).reshardSplitShardCountSummary()
+                );
             }
         }
 
@@ -642,12 +657,21 @@ public void testOperationRoutingWithResharding() throws IOException {
             searchShardsWithOneShardSplitAndRouting.get(indexOfShardWithNewState).iterator().shardId().id()
         );
         // Since the target shard is in SPLIT state, reshardSplitShardCountSummary is updated for it and the corresponding source shard.
-        assertEquals(newShardCount, searchShardsWithOneShardSplitAndRouting.get(indexOfShardWithNewState).reshardSplitShardCountSummary());
+        assertEquals(
+            SplitShardCountSummary.fromInt(newShardCount),
+            searchShardsWithOneShardSplitAndRouting.get(indexOfShardWithNewState).reshardSplitShardCountSummary()
+        );
         for (int i = 0; i < shardCount; i++) {
             if (i == sourceShard) {
-                assertEquals(newShardCount, searchShardsWithOneShardSplitAndRouting.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(newShardCount),
+                    searchShardsWithOneShardSplitAndRouting.get(i).reshardSplitShardCountSummary()
+                );
             } else {
-                assertEquals(shardCount, searchShardsWithOneShardSplitAndRouting.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(shardCount),
+                    searchShardsWithOneShardSplitAndRouting.get(i).reshardSplitShardCountSummary()
+                );
             }
         }
 
@@ -685,12 +709,21 @@ public void testOperationRoutingWithResharding() throws IOException {
         }
         assertEquals(shardChangingSplitTargetState, searchShardsWithOneShardDone.get(indexOfShardWithNewState).iterator().shardId().id());
         // Since the target shard is past SPLIT state, reshardSplitShardCountSummary is updated for it and the corresponding source shard.
-        assertEquals(newShardCount, searchShardsWithOneShardDone.get(indexOfShardWithNewState).reshardSplitShardCountSummary());
+        assertEquals(
+            SplitShardCountSummary.fromInt(newShardCount),
+            searchShardsWithOneShardDone.get(indexOfShardWithNewState).reshardSplitShardCountSummary()
+        );
         for (int i = 0; i < shardCount; i++) {
             if (i == sourceShard) {
-                assertEquals(newShardCount, searchShardsWithOneShardDone.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(newShardCount),
+                    searchShardsWithOneShardDone.get(i).reshardSplitShardCountSummary()
+                );
             } else {
-                assertEquals(shardCount, searchShardsWithOneShardDone.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(shardCount),
+                    searchShardsWithOneShardDone.get(i).reshardSplitShardCountSummary()
+                );
             }
         }
 
@@ -705,12 +738,21 @@ public void testOperationRoutingWithResharding() throws IOException {
             searchShardsWithOneShardDoneAndRouting.get(indexOfShardWithNewState).iterator().shardId().id()
         );
         // Since the target shard is past SPLIT state, reshardSplitShardCountSummary is updated for it and the corresponding source shard.
-        assertEquals(newShardCount, searchShardsWithOneShardDoneAndRouting.get(indexOfShardWithNewState).reshardSplitShardCountSummary());
+        assertEquals(
+            SplitShardCountSummary.fromInt(newShardCount),
+            searchShardsWithOneShardDoneAndRouting.get(indexOfShardWithNewState).reshardSplitShardCountSummary()
+        );
         for (int i = 0; i < shardCount; i++) {
             if (i == sourceShard) {
-                assertEquals(newShardCount, searchShardsWithOneShardDoneAndRouting.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(newShardCount),
+                    searchShardsWithOneShardDoneAndRouting.get(i).reshardSplitShardCountSummary()
+                );
             } else {
-                assertEquals(shardCount, searchShardsWithOneShardDoneAndRouting.get(i).reshardSplitShardCountSummary());
+                assertEquals(
+                    SplitShardCountSummary.fromInt(shardCount),
+                    searchShardsWithOneShardDoneAndRouting.get(i).reshardSplitShardCountSummary()
+                );
             }
         }