Handle Stale Requests (elastic#141301)

Handles requests (index/flush/refresh) that are "very" stale by rejecting them. This means requests that are more than 1 reshard operation behind. This includes stale requests that arrive after the resharding metadata is gone (even if a new split has not started).

Author: ankikuma

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -16,6 +16,7 @@
 import org.apache.lucene.store.LockObtainFailedException;
 import org.elasticsearch.action.bulk.IndexDocFailureStoreStatus;
 import org.elasticsearch.action.support.replication.ReplicationOperation;
+import org.elasticsearch.action.support.replication.StaleRequestException;
 import org.elasticsearch.cluster.RemoteException;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper;
@@ -78,6 +79,7 @@
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static java.util.Collections.unmodifiableMap;
+import static org.elasticsearch.action.support.replication.ReplicationSplitHelper.STALE_REQUEST_EXCEPTION_VERSION;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_UUID_NA_VALUE;
 import static org.elasticsearch.cluster.metadata.MetadataCreateIndexService.INDEX_LIMIT_EXCEEDED_EXCEPTION_VERSION;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
@@ -2049,7 +2051,8 @@ private enum ElasticsearchExceptionHandle {
             IndexLimitExceededException::new,
             186,
             INDEX_LIMIT_EXCEEDED_EXCEPTION_VERSION
-        );
+        ),
+        STALE_REQUEST_EXCEPTION(StaleRequestException.class, StaleRequestException::new, 187, STALE_REQUEST_EXCEPTION_VERSION);
 
         final Class<? extends ElasticsearchException> exceptionClass;
         final CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException> constructor;

server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java

@@ -12,10 +12,11 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.replication.ReplicationRequestSplitHelper;
+import org.elasticsearch.action.support.replication.BroadcastRequestSplitHelper;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -85,14 +86,20 @@ protected void shardOperationOnPrimary(
         }));
     }
 
-    // We are here because there was a mismatch between the SplitShardCountSummary in the request
-    // and that on the primary shard node. We assume that the request is exactly 1 reshard split behind
-    // the current state.
+    /**
+     * We are here because there was mismatch between the SplitShardCountSummary in the request
+     * and that on the primary shard node. In other words, the primary shard has moved ahead due to a split reshard
+     * operation after the request was created by the coordinator.
+     * We can assume that the request is exactly 1 reshard split behind the current state of the primary shard.
+     * This is because requests that are more than 1 reshard operation behind are rejected in
+     * {@link org.elasticsearch.action.support.replication.ReplicationSplitHelper
+     * #needsSplitCoordination(org.apache.logging.log4j.Logger, ReplicationRequest, IndexMetadata)}
+     */
     @Override
-    protected Map<ShardId, ShardFlushRequest> splitRequestOnPrimary(ShardFlushRequest request) {
-        return ReplicationRequestSplitHelper.splitRequest(
+    protected Map<ShardId, ShardFlushRequest> splitRequestOnPrimary(ShardFlushRequest request, ProjectMetadata project) {
+        return BroadcastRequestSplitHelper.splitRequest(
             request,
-            projectResolver.getProjectMetadata(clusterService.state()),
+            project,
             (targetShard, shardCountSummary) -> new ShardFlushRequest(request.getRequest(), targetShard, shardCountSummary)
         );
     }
@@ -103,7 +110,7 @@ protected Tuple<ReplicationResponse, Exception> combineSplitResponses(
         Map<ShardId, ShardFlushRequest> splitRequests,
         Map<ShardId, Tuple<ReplicationResponse, Exception>> responses
     ) {
-        return ReplicationRequestSplitHelper.combineSplitResponses(originalRequest, splitRequests, responses);
+        return BroadcastRequestSplitHelper.combineSplitResponses(originalRequest, splitRequests, responses);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java

@@ -15,11 +15,12 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.BasicReplicationRequest;
+import org.elasticsearch.action.support.replication.BroadcastRequestSplitHelper;
 import org.elasticsearch.action.support.replication.ReplicationOperation;
-import org.elasticsearch.action.support.replication.ReplicationRequestSplitHelper;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -112,14 +113,20 @@ protected void shardOperationOnPrimary(
         }));
     }
 
-    // We are here because there was mismatch between the SplitShardCountSummary in the request
-    // and that on the primary shard node. We assume that the request is exactly 1 reshard split behind
-    // the current state.
+    /**
+     * We are here because there was mismatch between the SplitShardCountSummary in the request
+     * and that on the primary shard node. In other words, the primary shard has moved ahead due to a split reshard
+     * operation after the request was created by the coordinator.
+     * We can assume that the request is exactly 1 reshard split behind the current state of the primary shard.
+     * This is because requests that are more than 1 reshard operation behind are rejected in
+     * {@link org.elasticsearch.action.support.replication.ReplicationSplitHelper
+     * #needsSplitCoordination(org.apache.logging.log4j.Logger, ReplicationRequest, IndexMetadata)}
+     */
     @Override
-    protected Map<ShardId, BasicReplicationRequest> splitRequestOnPrimary(BasicReplicationRequest request) {
-        return ReplicationRequestSplitHelper.splitRequest(
+    protected Map<ShardId, BasicReplicationRequest> splitRequestOnPrimary(BasicReplicationRequest request, ProjectMetadata project) {
+        return BroadcastRequestSplitHelper.splitRequest(
             request,
-            projectResolver.getProjectMetadata(clusterService.state()),
+            project,
             (targetShard, shardCountSummary) -> new BasicReplicationRequest(targetShard, shardCountSummary)
         );
     }
@@ -130,7 +137,7 @@ protected Tuple<ReplicationResponse, Exception> combineSplitResponses(
         Map<ShardId, BasicReplicationRequest> splitRequests,
         Map<ShardId, Tuple<ReplicationResponse, Exception>> responses
     ) {
-        return ReplicationRequestSplitHelper.combineSplitResponses(originalRequest, splitRequests, responses);
+        return BroadcastRequestSplitHelper.combineSplitResponses(originalRequest, splitRequests, responses);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.action.index.MappingUpdatedAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -165,8 +166,8 @@ protected void shardOperationOnPrimary(
     }
 
     @Override
-    protected Map<ShardId, BulkShardRequest> splitRequestOnPrimary(BulkShardRequest request) {
-        return ShardBulkSplitHelper.splitRequests(request, projectResolver.getProjectMetadata(clusterService.state()));
+    protected Map<ShardId, BulkShardRequest> splitRequestOnPrimary(BulkShardRequest request, ProjectMetadata project) {
+        return ShardBulkSplitHelper.splitRequests(request, project);
     }
 
     @Override

…ation/ReplicationRequestSplitHelper.java …ication/BroadcastRequestSplitHelper.javaserver/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequestSplitHelper.java renamed to server/src/main/java/org/elasticsearch/action/support/replication/BroadcastRequestSplitHelper.java server/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequestSplitHelper.java renamed to server/src/main/java/org/elasticsearch/action/support/replication/BroadcastRequestSplitHelper.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.support.replication;
 
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.routing.SplitShardCountSummary;
@@ -22,8 +23,17 @@
 import java.util.Map;
 import java.util.function.BiFunction;
 
-public final class ReplicationRequestSplitHelper {
-    private ReplicationRequestSplitHelper() {}
+/**
+ * This class implements the logic to "split", rather forward, Broadcast requests like refresh and flush
+ * to the target nodes after a Resharding operation. It also implements the logic to combine responses from
+ * source and target shards.
+ *
+ * The logic to split indexing requests is different because each document id has to be inspected to route it
+ * to the correct shard post Resharding. That logic is implemented in
+ * {@link org.elasticsearch.action.bulk.ShardBulkSplitHelper}
+ */
+public final class BroadcastRequestSplitHelper {
+    private BroadcastRequestSplitHelper() {}
 
     /**
      * Given a stale Replication Request, like flush or refresh, split it into multiple requests,
@@ -32,6 +42,11 @@ private ReplicationRequestSplitHelper() {}
      * {@link org.elasticsearch.action.bulk.BulkShardRequest}
      * We are here because there was a mismatch between the SplitShardCountSummary in the request
      * and that on the primary shard node.
+     *
+     * Note that {@link org.elasticsearch.cluster.metadata.IndexReshardingMetadata} cannot be NULL here
+     * because it is evaluated in {@link ReplicationSplitHelper#needsSplitCoordination(Logger, ReplicationRequest, IndexMetadata)}
+     * with the same metadata before arriving here. The request would have been rejected there if there was no Resharding metadata found.
+     *
      * TODO:
      * We assume here that the request is exactly 1 reshard split behind
      * the current state. We might either revise this assumption or enforce it
@@ -52,7 +67,7 @@ public static <T extends ReplicationRequest<T>> Map<ShardId, T> splitRequest(
         // Create a request for original source shard and for each target shard.
         // New requests that are to be handled by target shards should contain the
         // latest ShardCountSummary.
-        // TODO: This will not work if the reshard metadata is gone
+        assert indexMetadata.getReshardingMetadata() != null;
         int targetShardId = indexMetadata.getReshardingMetadata().getSplit().targetShard(sourceShard.id());
         ShardId targetShard = new ShardId(sourceShard.getIndex(), targetShardId);
 

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

@@ -11,10 +11,12 @@
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.RetryableAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexReshardingMetadata;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -42,6 +44,7 @@ public class ReplicationSplitHelper<
     ReplicaRequest extends ReplicationRequest<ReplicaRequest>,
     Response extends ReplicationResponse> {
 
+    public static TransportVersion STALE_REQUEST_EXCEPTION_VERSION = TransportVersion.fromName("stale_request_exception");
     private final Logger logger;
     private final ClusterService clusterService;
     private final TimeValue initialRetryBackoffBound;
@@ -66,14 +69,38 @@ public ReplicationSplitHelper(
     }
 
     public static <Request extends ReplicationRequest<Request>> boolean needsSplitCoordination(
+        final Logger logger,
         final Request primaryRequest,
         final IndexMetadata indexMetadata
-    ) {
+    ) throws Exception {
         SplitShardCountSummary requestSplitSummary = primaryRequest.reshardSplitShardCountSummary();
-        // TODO: We currently only set the request split summary transport shard bulk. Only evaluate this at the moment or else every
+        // TODO: We currently only set the request split summary for certain Replication Requests
+        // like refresh, flush and shard bulk requests. Only evaluate this when set or else every
         // request would say it needs a split.
-        return requestSplitSummary.isUnset() == false
-            && requestSplitSummary.equals(SplitShardCountSummary.forIndexing(indexMetadata, primaryRequest.shardId().getId())) == false;
+        if (requestSplitSummary.isUnset()) { // no split coordination required
+            return false;
+        }
+
+        SplitShardCountSummary latestSplitSummary = SplitShardCountSummary.forIndexing(indexMetadata, primaryRequest.shardId().getId());
+        if (requestSplitSummary.equals(latestSplitSummary)) {  // no split coordination required
+            return false;
+        } else {  // check that resharding is ongoing and the latest shard count is exactly 2 times the shard count in the request
+            if (indexMetadata.getReshardingMetadata() == null
+                || latestSplitSummary.asInt() != IndexReshardingMetadata.RESHARD_SPLIT_FACTOR * requestSplitSummary.asInt()) {
+                if (indexMetadata.getReshardingMetadata() == null) {
+                    logger.debug("Request is stale, expected resharding metadata but none found");
+                } else {
+                    logger.debug(
+                        "Request is stale due to concurrent reshard operation, expected shardCountSummary [{}] but found [{}]",
+                        requestSplitSummary.asInt(),
+                        latestSplitSummary.asInt()
+                    );
+                }
+                throw new StaleRequestException(primaryRequest.index());
+            }
+            return true;
+        }
+
     }
 
     @FunctionalInterface
@@ -137,7 +164,7 @@ public SplitCoordinator(
         }
 
         public void coordinate() throws Exception {
-            Map<ShardId, Request> splitRequests = action.splitRequestOnPrimary(originalRequest);
+            Map<ShardId, Request> splitRequests = action.splitRequestOnPrimary(originalRequest, project);
 
             int numSplitRequests = splitRequests.size();
 

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.support.replication;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.rest.RestStatus;
+
+import java.io.IOException;
+
+/**
+ * An exception indicating a stale request during resharding.
+ */
+public class StaleRequestException extends ElasticsearchException {
+
+    public StaleRequestException(String index) {
+        super("Request for index [{}] is stale due to concurrent reshard operation, retry after sometime", index);
+    }
+
+    public StaleRequestException(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public final RestStatus status() {
+        return RestStatus.SERVICE_UNAVAILABLE;
+    }
+}

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

@@ -339,10 +339,14 @@ protected abstract void shardOperationOnReplica(
     /**
      * During Resharding, we might need to split the primary request.
      * We are here because there was mismatch between the SplitShardCountSummary in the request
-     * and that on the primary shard node. We assume that the request is exactly 1 reshard split behind
-     * the current state.
+     * and that on the primary shard node. In other words, the primary shard has moved ahead due to a split reshard
+     * operation after the request was created by the coordinator.
+     * We can assume that the request is exactly 1 reshard split behind the current state of the primary shard.
+     * This is because requests that are more than 1 reshard operation behind are rejected in
+     * {@link org.elasticsearch.action.support.replication.ReplicationSplitHelper
+     * #needsSplitCoordination(org.apache.logging.log4j.Logger, ReplicationRequest, IndexMetadata)}
      */
-    protected Map<ShardId, Request> splitRequestOnPrimary(Request request) {
+    protected Map<ShardId, Request> splitRequestOnPrimary(Request request, ProjectMetadata project) {
         return Map.of(request.shardId(), request);
     }
 
@@ -533,7 +537,7 @@ void runWithPrimaryShardReference(final PrimaryShardReference primaryShardRefere
                             TransportResponseHandler.TRANSPORT_WORKER
                         )
                     );
-                } else if (ReplicationSplitHelper.needsSplitCoordination(primaryRequest.getRequest(), indexMetadata)) {
+                } else if (ReplicationSplitHelper.needsSplitCoordination(logger, primaryRequest.getRequest(), indexMetadata)) {
                     ReplicationSplitHelper<Request, ReplicaRequest, Response>.SplitCoordinator splitCoordinator = splitHelper
                         .newSplitRequest(
                             TransportReplicationAction.this,

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

@@ -91,6 +91,11 @@
 public class IndexReshardingMetadata implements ToXContentFragment, Writeable {
     private static final String SPLIT_FIELD_NAME = "split";
     private static final ParseField SPLIT_FIELD = new ParseField(SPLIT_FIELD_NAME);
+
+    // During a reshard split operation, we restrict the split factor to be exactly 2 i.e
+    // we can go from 1 -> 2 shards, or 2 -> 4 shards, and so on.
+    public static final int RESHARD_SPLIT_FACTOR = 2;
+
     // This exists only so that tests can verify that IndexReshardingMetadata supports more than one kind of operation.
     // It can be removed when we have defined a second real operation, such as shrink.
     private static final String NOOP_FIELD_NAME = "noop";

server/src/main/resources/transport/definitions/referable/stale_request_exception.csv

@@ -0,0 +1 @@
+9274000