Simplify write failure handling (backport of #19105) (#22778)

* Simplify write failure handling (backport of #19105)

Currently, any write (e.g. `index`, `delete`) operation failure can be categorized as:
- request failure (e.g. analysis, parsing error, version conflict)
- transient operation failure (e.g. due to shard initializing, relocation)
- environment failure (e.g. out of disk, corruption, lucene tragic event)

The main motivation of the PR is to handle these failure types appropriately for a
write request. Each failure type needs to be handled differently:
- request failure (being request specific) should be replicated and then failed
- transient failure should be retried (eventually succeeding)
- environment failure (persistent primary shard failure) should fail the request
  immediately.

Currently, transient operation failures are retried in replication action but no distinction
is made between request and environment failures, both fails write request immediately.

In this PR, we distinguish between request and environment failures for a write operation.
In case of environment failures, the exception is bubbled up failing the request and in case
of request failures, the exception is captured and replication continues (we ignore performing
on replicas when such failures occur in primary). Transient operation failures are bubbled up
to be retried by the replication operation, as before.

* incorporate feedback

Author: areek

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

@@ -40,7 +40,6 @@
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
-import org.elasticsearch.index.translog.Translog.Location;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -49,7 +48,7 @@
 /**
  * Performs the delete operation.
  */
-public class TransportDeleteAction extends TransportWriteAction<DeleteRequest, DeleteResponse> {
+public class TransportDeleteAction extends TransportWriteAction<DeleteRequest, DeleteRequest,DeleteResponse> {
 
     private final AutoCreateIndex autoCreateIndex;
     private final TransportCreateIndexAction createIndexAction;
@@ -61,7 +60,7 @@ public TransportDeleteAction(Settings settings, TransportService transportServic
                                  IndexNameExpressionResolver indexNameExpressionResolver,
                                  AutoCreateIndex autoCreateIndex) {
         super(settings, DeleteAction.NAME, transportService, clusterService, indicesService, threadPool, shardStateAction, actionFilters,
-                indexNameExpressionResolver, DeleteRequest::new, ThreadPool.Names.INDEX);
+                indexNameExpressionResolver, DeleteRequest::new, DeleteRequest::new, ThreadPool.Names.INDEX);
         this.createIndexAction = createIndexAction;
         this.autoCreateIndex = autoCreateIndex;
     }
@@ -70,7 +69,11 @@ public TransportDeleteAction(Settings settings, TransportService transportServic
     protected void doExecute(Task task, final DeleteRequest request, final ActionListener<DeleteResponse> listener) {
         ClusterState state = clusterService.state();
         if (autoCreateIndex.shouldAutoCreate(request.index(), state)) {
-            createIndexAction.execute(task, new CreateIndexRequest().index(request.index()).cause("auto(delete api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() {
+            CreateIndexRequest createIndexRequest = new CreateIndexRequest()
+                    .index(request.index())
+                    .cause("auto(delete api)")
+                    .masterNodeTimeout(request.timeout());
+            createIndexAction.execute(task, createIndexRequest, new ActionListener<CreateIndexResponse>() {
                 @Override
                 public void onResponse(CreateIndexResponse result) {
                     innerExecute(task, request, listener);
@@ -119,30 +122,33 @@ protected DeleteResponse newResponseInstance() {
     }
 
     @Override
-    protected WriteResult<DeleteResponse> onPrimaryShard(DeleteRequest request, IndexShard indexShard) {
-        return executeDeleteRequestOnPrimary(request, indexShard);
+    protected WritePrimaryResult shardOperationOnPrimary(DeleteRequest request, IndexShard primary) throws Exception {
+        final Engine.DeleteResult result = executeDeleteRequestOnPrimary(request, primary);
+        final DeleteResponse response = result.hasFailure() ? null :
+                new DeleteResponse(primary.shardId(), request.type(), request.id(), result.getVersion(), result.isFound());
+        return new WritePrimaryResult(request, response, result.getTranslogLocation(), result.getFailure(), primary);
     }
 
     @Override
-    protected Location onReplicaShard(DeleteRequest request, IndexShard indexShard) {
-        return executeDeleteRequestOnReplica(request, indexShard).getTranslogLocation();
+    protected WriteReplicaResult shardOperationOnReplica(DeleteRequest request, IndexShard replica) throws Exception {
+        final Engine.DeleteResult result = executeDeleteRequestOnReplica(request, replica);
+        return new WriteReplicaResult(request, result.getTranslogLocation(), result.getFailure(), replica);
     }
 
-    public static WriteResult<DeleteResponse> executeDeleteRequestOnPrimary(DeleteRequest request, IndexShard indexShard) {
-        Engine.Delete delete = indexShard.prepareDeleteOnPrimary(request.type(), request.id(), request.version(), request.versionType());
-        indexShard.delete(delete);
-        // update the request with the version so it will go to the replicas
-        request.versionType(delete.versionType().versionTypeForReplicationAndRecovery());
-        request.version(delete.version());
-
-        assert request.versionType().validateVersionForWrites(request.version());
-        DeleteResponse response = new DeleteResponse(indexShard.shardId(), request.type(), request.id(), delete.version(), delete.found());
-        return new WriteResult<>(response, delete.getTranslogLocation());
+    public static Engine.DeleteResult executeDeleteRequestOnPrimary(DeleteRequest request, IndexShard primary) {
+        Engine.Delete delete = primary.prepareDeleteOnPrimary(request.type(), request.id(), request.version(), request.versionType());
+        Engine.DeleteResult result = primary.delete(delete);
+        if (result.hasFailure() == false) {
+            // update the request with the version so it will go to the replicas
+            request.versionType(delete.versionType().versionTypeForReplicationAndRecovery());
+            request.version(result.getVersion());
+            assert request.versionType().validateVersionForWrites(request.version());
+        }
+        return result;
     }
 
-    public static Engine.Delete executeDeleteRequestOnReplica(DeleteRequest request, IndexShard indexShard) {
-        Engine.Delete delete = indexShard.prepareDeleteOnReplica(request.type(), request.id(), request.version(), request.versionType());
-        indexShard.delete(delete);
-        return delete;
+    public static Engine.DeleteResult executeDeleteRequestOnReplica(DeleteRequest request, IndexShard replica) {
+        Engine.Delete delete = replica.prepareDeleteOnReplica(request.type(), request.id(), request.version(), request.versionType());
+        return replica.delete(delete);
     }
 }

core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java

@@ -44,11 +44,11 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 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.Location;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.tasks.Task;
@@ -65,7 +65,7 @@
  * <li><b>allowIdGeneration</b>: If the id is set not, should it be generated. Defaults to <tt>true</tt>.
  * </ul>
  */
-public class TransportIndexAction extends TransportWriteAction<IndexRequest, IndexResponse> {
+public class TransportIndexAction extends TransportWriteAction<IndexRequest, IndexRequest, IndexResponse> {
 
     private final AutoCreateIndex autoCreateIndex;
     private final boolean allowIdGeneration;
@@ -83,7 +83,7 @@ public TransportIndexAction(Settings settings, TransportService transportService
                                 MappingUpdatedAction mappingUpdatedAction, ActionFilters actionFilters,
                                 IndexNameExpressionResolver indexNameExpressionResolver, AutoCreateIndex autoCreateIndex) {
         super(settings, IndexAction.NAME, transportService, clusterService, indicesService, threadPool, shardStateAction,
-            actionFilters, indexNameExpressionResolver, IndexRequest::new, ThreadPool.Names.INDEX);
+            actionFilters, indexNameExpressionResolver, IndexRequest::new, IndexRequest::new, ThreadPool.Names.INDEX);
         this.mappingUpdatedAction = mappingUpdatedAction;
         this.createIndexAction = createIndexAction;
         this.autoCreateIndex = autoCreateIndex;
@@ -162,65 +162,88 @@ protected IndexResponse newResponseInstance() {
     }
 
     @Override
-    protected WriteResult<IndexResponse> onPrimaryShard(IndexRequest request, IndexShard indexShard) throws Exception {
-        return executeIndexRequestOnPrimary(request, indexShard, mappingUpdatedAction);
+    protected WritePrimaryResult shardOperationOnPrimary(IndexRequest request, IndexShard primary) throws Exception {
+        final Engine.IndexResult indexResult = executeIndexRequestOnPrimary(request, primary, mappingUpdatedAction);
+        final IndexResponse response = indexResult.hasFailure() ? null :
+                new IndexResponse(primary.shardId(), request.type(), request.id(), indexResult.getVersion(),
+                        indexResult.isCreated());
+        return new WritePrimaryResult(request, response, indexResult.getTranslogLocation(), indexResult.getFailure(), primary);
     }
 
     @Override
-    protected Location onReplicaShard(IndexRequest request, IndexShard indexShard) {
-        return executeIndexRequestOnReplica(request, indexShard).getTranslogLocation();
+    protected WriteReplicaResult shardOperationOnReplica(IndexRequest request, IndexShard replica) throws Exception {
+        final Engine.IndexResult indexResult = executeIndexRequestOnReplica(request, replica);
+        return new WriteReplicaResult(request, indexResult.getTranslogLocation(), indexResult.getFailure(), replica);
     }
 
     /**
      * Execute the given {@link IndexRequest} on a replica shard, throwing a
      * {@link RetryOnReplicaException} if the operation needs to be re-tried.
      */
-    public static Engine.Index executeIndexRequestOnReplica(IndexRequest request, IndexShard indexShard) {
-        final ShardId shardId = indexShard.shardId();
+    public static Engine.IndexResult executeIndexRequestOnReplica(IndexRequest request, IndexShard replica) {
+        final ShardId shardId = replica.shardId();
         SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.REPLICA, shardId.getIndexName(), request.type(), request.id(), request.source())
                 .routing(request.routing()).parent(request.parent()).timestamp(request.timestamp()).ttl(request.ttl());
 
-        final Engine.Index operation = indexShard.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
+        final Engine.Index operation;
+        try {
+            operation = replica.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
+        } catch (MapperParsingException e) {
+            return new Engine.IndexResult(e, request.version());
+        }
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
         if (update != null) {
             throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
         }
-        indexShard.index(operation);
-        return operation;
+        return replica.index(operation);
     }
 
     /** Utility method to prepare an index operation on primary shards */
-    public static Engine.Index prepareIndexOperationOnPrimary(IndexRequest request, IndexShard indexShard) {
+    static Engine.Index prepareIndexOperationOnPrimary(IndexRequest request, IndexShard primary) {
         SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.PRIMARY, request.index(), request.type(), request.id(), request.source())
             .routing(request.routing()).parent(request.parent()).timestamp(request.timestamp()).ttl(request.ttl());
-        return indexShard.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
+        return primary.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
     }
 
-    public static WriteResult<IndexResponse> executeIndexRequestOnPrimary(IndexRequest request, IndexShard indexShard,
+    public static Engine.IndexResult executeIndexRequestOnPrimary(IndexRequest request, IndexShard primary,
             MappingUpdatedAction mappingUpdatedAction) throws Exception {
-        Engine.Index operation = prepareIndexOperationOnPrimary(request, indexShard);
+        Engine.Index operation;
+        try {
+            operation = prepareIndexOperationOnPrimary(request, primary);
+        } catch (MapperParsingException | IllegalArgumentException e) {
+            return new Engine.IndexResult(e, request.version());
+        }
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
-        final ShardId shardId = indexShard.shardId();
+        final ShardId shardId = primary.shardId();
         if (update != null) {
-            mappingUpdatedAction.updateMappingOnMaster(shardId.getIndex(), request.type(), update);
-            operation = prepareIndexOperationOnPrimary(request, indexShard);
+            // can throw timeout exception when updating mappings or ISE for attempting to update default mappings
+            // which are bubbled up
+            try {
+                mappingUpdatedAction.updateMappingOnMaster(shardId.getIndex(), request.type(), update);
+            } catch (IllegalArgumentException e) {
+                // throws IAE on conflicts merging dynamic mappings
+                return new Engine.IndexResult(e, request.version());
+            }
+            try {
+                operation = prepareIndexOperationOnPrimary(request, primary);
+            } catch (MapperParsingException | IllegalArgumentException e) {
+                return new Engine.IndexResult(e, request.version());
+            }
             update = operation.parsedDoc().dynamicMappingsUpdate();
             if (update != null) {
                 throw new ReplicationOperation.RetryOnPrimaryException(shardId,
                     "Dynamic mappings are not available on the node that holds the primary yet");
             }
         }
-        indexShard.index(operation);
-
-        // update the version on request so it will happen on the replicas
-        final long version = operation.version();
-        request.version(version);
-        request.versionType(request.versionType().versionTypeForReplicationAndRecovery());
-
-        assert request.versionType().validateVersionForWrites(request.version());
-
-        IndexResponse response = new IndexResponse(shardId, request.type(), request.id(), request.version(), operation.isCreated());
-        return new WriteResult<>(response, operation.getTranslogLocation());
+        Engine.IndexResult result = primary.index(operation);
+        if (result.hasFailure() == false) {
+            // update the version on request so it will happen on the replicas
+            final long version = result.getVersion();
+            request.version(version);
+            request.versionType(request.versionType().versionTypeForReplicationAndRecovery());
+            assert request.versionType().validateVersionForWrites(request.version());
+        }
+        return result;
     }
 
     private void processIngestIndexRequest(Task task, IndexRequest indexRequest, ActionListener listener) {

core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
@@ -112,21 +113,23 @@ public void execute() throws Exception {
         pendingActions.incrementAndGet();
         primaryResult = primary.perform(request);
         final ReplicaRequest replicaRequest = primaryResult.replicaRequest();
-        assert replicaRequest.primaryTerm() > 0 : "replicaRequest doesn't have a primary term";
-        if (logger.isTraceEnabled()) {
-            logger.trace("[{}] op [{}] completed on primary for request [{}]", primaryId, opType, request);
-        }
+        if (replicaRequest != null) {
+            assert replicaRequest.primaryTerm() > 0 : "replicaRequest doesn't have a primary term";
+            if (logger.isTraceEnabled()) {
+                logger.trace("[{}] op [{}] completed on primary for request [{}]", primaryId, opType, request);
+            }
 
-        // we have to get a new state after successfully indexing into the primary in order to honour recovery semantics.
-        // we have to make sure that every operation indexed into the primary after recovery start will also be replicated
-        // to the recovery target. If we use an old cluster state, we may miss a relocation that has started since then.
-        ClusterState clusterState = clusterStateSupplier.get();
-        final List<ShardRouting> shards = getShards(primaryId, clusterState);
-        Set<String> inSyncAllocationIds = getInSyncAllocationIds(primaryId, clusterState);
+            // we have to get a new state after successfully indexing into the primary in order to honour recovery semantics.
+            // we have to make sure that every operation indexed into the primary after recovery start will also be replicated
+            // to the recovery target. If we use an old cluster state, we may miss a relocation that has started since then.
+            ClusterState clusterState = clusterStateSupplier.get();
+            final List<ShardRouting> shards = getShards(primaryId, clusterState);
+            Set<String> inSyncAllocationIds = getInSyncAllocationIds(primaryId, clusterState);
 
-        markUnavailableShardsAsStale(replicaRequest, inSyncAllocationIds, shards);
+            markUnavailableShardsAsStale(replicaRequest, inSyncAllocationIds, shards);
 
-        performOnReplicas(replicaRequest, shards);
+            performOnReplicas(replicaRequest, shards);
+        }
 
         successfulShards.incrementAndGet();
         decPendingAndFinishIfNeeded();
@@ -419,7 +422,11 @@ public RetryOnPrimaryException(StreamInput in) throws IOException {
 
     public interface PrimaryResult<R extends ReplicationRequest<R>> {
 
-        R replicaRequest();
+        /**
+         * @return null if no operation needs to be sent to a replica
+         * (for example when the operation failed on the primary due to a parsing exception)
+         */
+        @Nullable R replicaRequest();
 
         void setShardInfo(ReplicationResponse.ShardInfo shardInfo);
     }