Fail realtime get request if it is stale due to resharding (elastic#141604)

During resharding, a get request may become stale if the shard
that owns the requested document changes between the time that
the request is formed and when it is processed. We can detect
whether this has happened by comparing the current routing of
the document based on the routing table to the shard that has
actually processed the document. If there is a mismatch, we
fail the request.

Later this failure should be handled by internally retrying the
request with a fresh route, but for correctness we do need to
start by rejecting the request if it has been routed to the
wrong shard.

Author: bcully

server/src/main/java/org/elasticsearch/action/get/TransportGetFromTranslogAction.java

@@ -72,7 +72,8 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
                         getRequest.version(),
                         getRequest.versionType(),
                         getRequest.fetchSourceContext(),
-                        getRequest.isForceSyntheticSource()
+                        getRequest.isForceSyntheticSource(),
+                        getRequest.getSplitShardCountSummary()
                     );
                 long segmentGeneration = -1;
                 if (result == null) {

server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetFomTranslogAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.TransportActions;
+import org.elasticsearch.cluster.routing.SplitShardCountSummary;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -75,7 +76,8 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
                                 item.version(),
                                 item.versionType(),
                                 item.fetchSourceContext(),
-                                multiGetShardRequest.isForceSyntheticSource()
+                                multiGetShardRequest.isForceSyntheticSource(),
+                                SplitShardCountSummary.UNSET
                             );
                         GetResponse getResponse = null;
                         if (result == null) {

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

@@ -276,7 +276,8 @@ public int deleteShard(String id, @Nullable String routing) {
         @Override
         public int getShard(String id, @Nullable String routing) {
             checkRoutingRequired(id, routing);
-            return shardId(id, routing);
+            int shardId = shardId(id, routing);
+            return rerouteSearchIfResharding(shardId);
         }
 
         private void checkRoutingRequired(String id, @Nullable String routing) {

server/src/main/java/org/elasticsearch/index/get/ShardGetService.java

@@ -222,36 +222,35 @@ private GetResult doGet(
             } else {
                 missingMetric.inc(System.nanoTime() - now);
             }
-            if (getResult == null || getResult.isExists() == false) {
-                // during resharding, a coordinating node may route a get request to a shard that is the source of a split
-                // before the target shard has taken over that document, but by the time the request is processed on the
-                // source shard, handoff has occurred. If the get succeeds, this is fine - we block refreshes during this
-                // period so although indexing may have occured on the target shard, the source shard's copy is still valid
-                // because refresh hasn't happened. However, if the source shard doesn't have the document in this case, it
-                // may be that it is because the shard has deleted unowned documents after the split. Normally this shouldn't
-                // occur because we will delay deleting those documents for some grace period, but if it does happen we should
-                // fail the request rather than possibly incorrectly reporting that the document doesn't exist, when it may
-                // in fact be present on a different shard.
-                // We can defer this check until we know whether the response is missing, so that the common case doesn't
-                // have to resolve the current split shard summary. This way also means that a race between the get and handoff state
-                // can only cause us to throw an exception unnecessarily, rather than incorrectly returning a missing document.
-                final var indexMetadata = mapperService.getIndexSettings().getIndexMetadata();
-                final var currentSummary = SplitShardCountSummary.forSearch(indexMetadata, shardId().getId());
+            if (getResult == null || getResult.isExists() == false || realtime) {
                 if (splitShardCountSummary.equals(SplitShardCountSummary.UNSET)) {
                     // TODO, this should only be possible temporarily, until we've ensured that all callers provide a valid summary.
                     return getResult;
                 }
+                // during resharding, a coordinating node may route a get request to a shard that is the source of a split
+                // before the target shard has taken over that document, but by the time the request is processed on the
+                // source shard, handoff has occurred. If a non-realtime get succeeds, this is fine - we block refreshes during this
+                // period so although the document may have been updated on the target shard, the source shard's copy is still valid
+                // because refresh hasn't happened. However, if the get is a realtime get, or if the source shard doesn't have the
+                // document (perhaps because the shard has deleted unowned documents after the split) then the answer may be
+                // incorrect. So, if the request's split shard count summary indicates that routing has changed since the request
+                // was formulated, we double check that the requested document still maps to this shard.
+                final var indexMetadata = mapperService.getIndexSettings().getIndexMetadata();
+                // For realtime get, correct results depend on index routing (the new shard may accept updates at handoff) and consult
+                // the index shard's translog.
+                // Regular search happens on the search shard and uses search routing.
+                final var currentSummary = realtime
+                    ? SplitShardCountSummary.forIndexing(indexMetadata, shardId().getId())
+                    : SplitShardCountSummary.forSearch(indexMetadata, shardId().getId());
                 if (splitShardCountSummary.compareTo(currentSummary) >= 0) {
                     // coordinator is current, so response is valid
                     return getResult;
                 }
                 // Otherwise, recompute the route of the requested document based on current metadata and fail the request if it
-                // doesn't map to this shard anymore, since delete unowned may have removed it by this point. This is conservative:
-                // the document may genuinely not have existed, but unless we can be certain that delete-unowned hasn't run yet
-                // (which is difficult, because the index shard may see the target move to done before the search shard) it is safer
-                // to fail the request.
+                // doesn't map to this shard anymore.
                 final var indexRouting = IndexRouting.fromIndexMetadata(indexMetadata);
-                final var docShard = indexRouting.getShard(id, routing);
+                // see currentSummary above
+                final var docShard = realtime ? indexRouting.updateShard(id, routing) : indexRouting.getShard(id, routing);
                 if (docShard != shardId().getId()) {
                     // XXX we may want a more specific exception type here
                     throw new ElasticsearchStatusException("stale get request for document [" + id + "]", RestStatus.SERVICE_UNAVAILABLE);
@@ -272,7 +271,8 @@ public GetResult getFromTranslog(
         long version,
         VersionType versionType,
         FetchSourceContext fetchSourceContext,
-        boolean forceSyntheticSource
+        boolean forceSyntheticSource,
+        SplitShardCountSummary splitShardCountSummary
     ) throws IOException {
         return doGet(
             id,
@@ -285,7 +285,7 @@ public GetResult getFromTranslog(
             UNASSIGNED_PRIMARY_TERM,
             fetchSourceContext,
             forceSyntheticSource,
-            SplitShardCountSummary.UNSET,
+            splitShardCountSummary,
             indexShard::getFromTranslog
         );
     }

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

@@ -850,7 +850,7 @@ private static IndexRouting getSplitRouting(IndexMetadata startingMetadata) {
         );
     }
 
-    public void testCollectSearchShardsUnpartitionedWithResharding() throws IOException {
+    public void testCollectSearchShardsUnpartitionedAndGetWithResharding() {
         int shards = 1;
         int newShardCount = 2;
         var initialRouting = IndexRouting.fromIndexMetadata(
@@ -885,6 +885,7 @@ public void testCollectSearchShardsUnpartitionedWithResharding() throws IOExcept
             assertEquals(1, collectedShards.size());
             // Rerouting is in effect due to resharding metadata having a shard in CLONE state.
             assertEquals(0, collectedShards.get(0));
+            assertEquals(0, initialReshardingRouting.getShard(shardAndRouting.getValue(), null));
         }
 
         var reshardingRoutingWithShardInHandoff = IndexRouting.fromIndexMetadata(
@@ -903,8 +904,9 @@ public void testCollectSearchShardsUnpartitionedWithResharding() throws IOExcept
             var collectedShards = new ArrayList<>();
             reshardingRoutingWithShardInHandoff.collectSearchShards(shardAndRouting.getValue(), collectedShards::add);
             assertEquals(1, collectedShards.size());
-            // Rerouting is in effect due to resharding metadata having a shard in CLONE state.
+            // Rerouting is in effect due to resharding metadata having a shard in HANDOFF state.
             assertEquals(0, collectedShards.get(0));
+            assertEquals(0, reshardingRoutingWithShardInHandoff.getShard(shardAndRouting.getValue(), null));
         }
 
         var reshardingRoutingWithShardInSplit = IndexRouting.fromIndexMetadata(
@@ -926,6 +928,7 @@ public void testCollectSearchShardsUnpartitionedWithResharding() throws IOExcept
             assertEquals(1, collectedShards.size());
             // Rerouting is no longer in effect since resharding metadata has SPLIT state for this shard
             assertEquals(shardAndRouting.getKey(), collectedShards.get(0));
+            assertEquals(shardAndRouting.getKey().intValue(), reshardingRoutingWithShardInSplit.getShard(shardAndRouting.getValue(), null));
         }
     }
 

server/src/test/java/org/elasticsearch/index/shard/ShardGetServiceTests.java

@@ -363,7 +363,16 @@ public void testGetFromTranslog() throws IOException {
 
         // Issue a get that would enforce safe access mode and switches the maps from unsafe to safe
         var getResult = primary.getService()
-            .getFromTranslog("2", new String[] { "foo" }, true, 1, VersionType.INTERNAL, FetchSourceContext.FETCH_SOURCE, false);
+            .getFromTranslog(
+                "2",
+                new String[] { "foo" },
+                true,
+                1,
+                VersionType.INTERNAL,
+                FetchSourceContext.FETCH_SOURCE,
+                false,
+                SplitShardCountSummary.UNSET
+            );
         assertNull(getResult);
         var lastUnsafeGeneration = engine.getLastUnsafeSegmentGenerationForGets();
         // last unsafe generation is set to last committed gen after the refresh triggered by realtime get
@@ -387,7 +396,8 @@ public void testGetFromTranslog() throws IOException {
                 1,
                 VersionType.INTERNAL,
                 FetchSourceContext.FETCH_SOURCE,
-                false
+                false,
+                SplitShardCountSummary.UNSET
             );
         assertNull(getResult);
         // But normal get would still work!
@@ -413,11 +423,29 @@ public void testGetFromTranslog() throws IOException {
         indexDoc(primary, "1", "{\"foo\" : \"baz\"}", XContentType.JSON, "foobar");
         // The first get in safe mode, would trigger a refresh, since we need to start tracking translog locations in the live version map
         getResult = primary.getService()
-            .getFromTranslog("1", new String[] { "foo" }, true, 1, VersionType.INTERNAL, FetchSourceContext.FETCH_SOURCE, false);
+            .getFromTranslog(
+                "1",
+                new String[] { "foo" },
+                true,
+                1,
+                VersionType.INTERNAL,
+                FetchSourceContext.FETCH_SOURCE,
+                false,
+                SplitShardCountSummary.UNSET
+            );
         assertTrue(getResult.isExists());
         assertEquals(engine.getLastUnsafeSegmentGenerationForGets(), lastUnsafeGeneration);
         getResult = primary.getService()
-            .getFromTranslog("2", new String[] { "foo" }, true, 1, VersionType.INTERNAL, FetchSourceContext.FETCH_SOURCE, false);
+            .getFromTranslog(
+                "2",
+                new String[] { "foo" },
+                true,
+                1,
+                VersionType.INTERNAL,
+                FetchSourceContext.FETCH_SOURCE,
+                false,
+                SplitShardCountSummary.UNSET
+            );
         assertNull(getResult);
         assertEquals(engine.getLastUnsafeSegmentGenerationForGets(), lastUnsafeGeneration);
 
@@ -435,7 +463,16 @@ public void testGetFromTranslog() throws IOException {
         assertFalse(LiveVersionMapTestUtils.isSafeAccessRequired(map));
         assertTrue(LiveVersionMapTestUtils.isUnsafe(map));
         getResult = primary.getService()
-            .getFromTranslog("2", new String[] { "foo" }, true, 1, VersionType.INTERNAL, FetchSourceContext.FETCH_SOURCE, false);
+            .getFromTranslog(
+                "2",
+                new String[] { "foo" },
+                true,
+                1,
+                VersionType.INTERNAL,
+                FetchSourceContext.FETCH_SOURCE,
+                false,
+                SplitShardCountSummary.UNSET
+            );
         assertNull(getResult);
         var lastUnsafeGeneration2 = engine.getLastUnsafeSegmentGenerationForGets();
         assertTrue(lastUnsafeGeneration2 > lastUnsafeGeneration);