Bulk operation fail to replicate operations when a mapping update times out (#30244)

Starting with the refactoring in #22778 (released in 5.3) we may fail to properly replicate operation when a mapping update on master fails. If a bulk
operations needs a mapping update half way, it will send a request to the master before continuing 
to index the operations. If that request times out or isn't acked (i.e., even one node in the cluster 
didn't process it within 30s), we end up throwing the exception and aborting the entire bulk. This is 
a problem because all operations that were processed so far are not replicated any more to the 
replicas.  Although these operations were never "acked" to the user (we threw an error) it cause the 
local checkpoint on the replicas to lag (on 6.x) and the primary and replica to diverge. 

This PR does a couple of things:
1) Most importantly, treat *any* mapping update failure as a document level failure, meaning only 
    the relevant indexing operation will fail.
2) Removes the mapping update callbacks from `IndexShard.applyIndexOperationOnPrimary` and 
    similar methods for simpler execution. We don't use exceptions any more when a mapping 
    update was successful.

I think we need to do more work here (the fact that a single slow node can prevent those mappings 
updates from being acked and thus fail operations is bad), but I want to keep this as small as I can 
(it is already too big).

Author: bleskes

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

@@ -29,11 +29,4 @@ public interface MappingUpdatePerformer {
      */
     void updateMappings(Mapping update, ShardId shardId, String type);
 
-    /**
-     *  Throws a {@code ReplicationOperation.RetryOnPrimaryException} if the operation needs to be
-     * retried on the primary due to the mappings not being present yet, or a different exception if
-     * updating the mappings on the master failed.
-     */
-    void verifyMappings(Mapping update, ShardId shardId);
-
 }

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

@@ -122,11 +122,11 @@ public static Translog.Location performOnReplica(ResyncReplicationRequest reques
         Translog.Location location = null;
         for (Translog.Operation operation : request.getOperations()) {
             try {
-                final Engine.Result operationResult = replica.applyTranslogOperation(operation, Engine.Operation.Origin.REPLICA,
-                    update -> {
-                        throw new TransportReplicationAction.RetryOnReplicaException(replica.shardId(),
-                            "Mappings are not available on the replica yet, triggered update: " + update);
-                    });
+                final Engine.Result operationResult = replica.applyTranslogOperation(operation, Engine.Operation.Origin.REPLICA);
+                if (operationResult.getResultType() == Engine.Result.Type.MAPPING_UPDATE_REQUIRED) {
+                    throw new TransportReplicationAction.RetryOnReplicaException(replica.shardId(),
+                        "Mappings are not available on the replica yet, triggered update: " + operationResult.getRequiredMappingUpdate());
+                }
                 location = syncOperationResultOrThrow(operationResult, location);
             } catch (Exception e) {
                 // if its not a failure to be ignored, let it bubble up

server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java

@@ -33,23 +33,17 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.mapper.MapperParsingException;
-import org.elasticsearch.index.mapper.Mapping;
-import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.Translog.Location;
 import org.elasticsearch.indices.IndicesService;
-import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -78,7 +72,7 @@ protected TransportWriteAction(Settings settings, String actionName, TransportSe
     protected static Location syncOperationResultOrThrow(final Engine.Result operationResult,
                                                          final Location currentLocation) throws Exception {
         final Location location;
-        if (operationResult.hasFailure()) {
+        if (operationResult.getFailure() != null) {
             // check if any transient write operation failures should be bubbled up
             Exception failure = operationResult.getFailure();
             assert failure instanceof MapperParsingException : "expected mapper parsing failures. got " + failure;

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

@@ -144,7 +144,7 @@ private void setReformat(boolean reformat) {
 
     @Override
     public void postIndex(ShardId shardId, Engine.Index indexOperation, Engine.IndexResult result) {
-        if (result.hasFailure() == false) {
+        if (result.getResultType() == Engine.Result.Type.SUCCESS) {
             final ParsedDocument doc = indexOperation.parsedDoc();
             final long tookInNanos = result.getTook();
             if (indexWarnThreshold >= 0 && tookInNanos > indexWarnThreshold) {

server/src/main/java/org/elasticsearch/index/IndexingSlowLog.java

@@ -58,6 +58,7 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
 import org.elasticsearch.index.VersionType;
+import org.elasticsearch.index.mapper.Mapping;
 import org.elasticsearch.index.mapper.ParseContext.Document;
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.merge.MergeStats;
@@ -295,27 +296,45 @@ public Condition newCondition() {
      **/
     public abstract static class Result {
         private final Operation.TYPE operationType;
+        private final Result.Type resultType;
         private final long version;
         private final long seqNo;
         private final Exception failure;
         private final SetOnce<Boolean> freeze = new SetOnce<>();
+        private final Mapping requiredMappingUpdate;
         private Translog.Location translogLocation;
         private long took;
 
         protected Result(Operation.TYPE operationType, Exception failure, long version, long seqNo) {
             this.operationType = operationType;
-            this.failure = failure;
+            this.failure = Objects.requireNonNull(failure);
             this.version = version;
             this.seqNo = seqNo;
+            this.requiredMappingUpdate = null;
+            this.resultType = Type.FAILURE;
         }
 
         protected Result(Operation.TYPE operationType, long version, long seqNo) {
-            this(operationType, null, version, seqNo);
+            this.operationType = operationType;
+            this.version = version;
+            this.seqNo = seqNo;
+            this.failure = null;
+            this.requiredMappingUpdate = null;
+            this.resultType = Type.SUCCESS;
+        }
+
+        protected Result(Operation.TYPE operationType, Mapping requiredMappingUpdate) {
+            this.operationType = operationType;
+            this.version = Versions.NOT_FOUND;
+            this.seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
+            this.failure = null;
+            this.requiredMappingUpdate = requiredMappingUpdate;
+            this.resultType = Type.MAPPING_UPDATE_REQUIRED;
         }
 
-        /** whether the operation had failure */
-        public boolean hasFailure() {
-            return failure != null;
+        /** whether the operation was successful, has failed or was aborted due to a mapping update */
+        public Type getResultType() {
+            return resultType;
         }
 
         /** get the updated document version */
@@ -332,6 +351,14 @@ public long getSeqNo() {
             return seqNo;
         }
 
+        /**
+         * If the operation was aborted due to missing mappings, this method will return the mappings
+         * that are required to complete the operation.
+         */
+        public Mapping getRequiredMappingUpdate() {
+            return requiredMappingUpdate;
+        }
+
         /** get the translog location after executing the operation */
         public Translog.Location getTranslogLocation() {
             return translogLocation;
@@ -371,6 +398,11 @@ void freeze() {
             freeze.set(true);
         }
 
+        public enum Type {
+            SUCCESS,
+            FAILURE,
+            MAPPING_UPDATE_REQUIRED
+        }
     }
 
     public static class IndexResult extends Result {
@@ -383,9 +415,8 @@ public IndexResult(long version, long seqNo, boolean created) {
         }
 
         /**
-         * use in case of index operation failed before getting to internal engine
-         * (e.g while preparing operation or updating mappings)
-         * */
+         * use in case of the index operation failed before getting to internal engine
+         **/
         public IndexResult(Exception failure, long version) {
             this(failure, version, SequenceNumbers.UNASSIGNED_SEQ_NO);
         }
@@ -395,6 +426,11 @@ public IndexResult(Exception failure, long version, long seqNo) {
             this.created = false;
         }
 
+        public IndexResult(Mapping requiredMappingUpdate) {
+            super(Operation.TYPE.INDEX, requiredMappingUpdate);
+            this.created = false;
+        }
+
         public boolean isCreated() {
             return created;
         }
@@ -410,11 +446,23 @@ public DeleteResult(long version, long seqNo, boolean found) {
             this.found = found;
         }
 
+        /**
+         * use in case of the delete operation failed before getting to internal engine
+         **/
+        public DeleteResult(Exception failure, long version) {
+            this(failure, version, SequenceNumbers.UNASSIGNED_SEQ_NO, false);
+        }
+
         public DeleteResult(Exception failure, long version, long seqNo, boolean found) {
             super(Operation.TYPE.DELETE, failure, version, seqNo);
             this.found = found;
         }
 
+        public DeleteResult(Mapping requiredMappingUpdate) {
+            super(Operation.TYPE.DELETE, requiredMappingUpdate);
+            this.found = false;
+        }
+
         public boolean isFound() {
             return found;
         }

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -765,7 +765,7 @@ public IndexResult index(Index index) throws IOException {
                 final IndexResult indexResult;
                 if (plan.earlyResultOnPreFlightError.isPresent()) {
                     indexResult = plan.earlyResultOnPreFlightError.get();
-                    assert indexResult.hasFailure();
+                    assert indexResult.getResultType() == Result.Type.FAILURE : indexResult.getResultType();
                 } else if (plan.indexIntoLucene) {
                     indexResult = indexIntoLucene(index, plan);
                 } else {
@@ -774,7 +774,7 @@ public IndexResult index(Index index) throws IOException {
                 }
                 if (index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
                     final Translog.Location location;
-                    if (indexResult.hasFailure() == false) {
+                    if (indexResult.getResultType() == Result.Type.SUCCESS) {
                         location = translog.add(new Translog.Index(index, indexResult));
                     } else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                         // if we have document failure, record it as a no-op in the translog with the generated seq_no
@@ -784,7 +784,7 @@ public IndexResult index(Index index) throws IOException {
                     }
                     indexResult.setTranslogLocation(location);
                 }
-                if (plan.indexIntoLucene && indexResult.hasFailure() == false) {
+                if (plan.indexIntoLucene && indexResult.getResultType() == Result.Type.SUCCESS) {
                     final Translog.Location translogLocation = trackTranslogLocation.get() ? indexResult.getTranslogLocation() : null;
                     versionMap.maybePutIndexUnderLock(index.uid().bytes(),
                         new IndexVersionValue(translogLocation, plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm()));
@@ -1087,7 +1087,7 @@ public DeleteResult delete(Delete delete) throws IOException {
             }
             if (delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
                 final Translog.Location location;
-                if (deleteResult.hasFailure() == false) {
+                if (deleteResult.getResultType() == Result.Type.SUCCESS) {
                     location = translog.add(new Translog.Delete(delete, deleteResult));
                 } else if (deleteResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                     location = translog.add(new Translog.NoOp(deleteResult.getSeqNo(),