Make index and delete operation execute as single bulk item (backport of #21964)

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

@@ -22,8 +22,8 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.Version;
-import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.DocWriteRequest.OpType;
+import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.update.UpdateResponse;
@@ -92,10 +92,10 @@ public static class Failure implements Writeable, ToXContent {
         private final String index;
         private final String type;
         private final String id;
-        private final Throwable cause;
+        private final Exception cause;
         private final RestStatus status;
 
-        public Failure(String index, String type, String id, Throwable cause) {
+        public Failure(String index, String type, String id, Exception cause) {
             this.index = index;
             this.type = type;
             this.id = id;
@@ -161,7 +161,7 @@ public RestStatus getStatus() {
         /**
          * The actual cause of the failure.
          */
-        public Throwable getCause() {
+        public Exception getCause() {
             return cause;
         }
 

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.action.support.replication.ReplicationResponse.ShardInfo;
 import org.elasticsearch.action.support.replication.TransportWriteAction;
@@ -50,7 +51,10 @@
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 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.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -59,10 +63,6 @@
 
 import java.util.Map;
 
-import static org.elasticsearch.action.delete.TransportDeleteAction.executeDeleteRequestOnPrimary;
-import static org.elasticsearch.action.delete.TransportDeleteAction.executeDeleteRequestOnReplica;
-import static org.elasticsearch.action.index.TransportIndexAction.executeIndexRequestOnPrimary;
-import static org.elasticsearch.action.index.TransportIndexAction.executeIndexRequestOnReplica;
 import static org.elasticsearch.action.support.replication.ReplicationOperation.ignoreReplicaException;
 import static org.elasticsearch.action.support.replication.ReplicationOperation.isConflictException;
 
@@ -103,7 +103,8 @@ protected boolean resolveIndex() {
     }
 
     @Override
-    protected WritePrimaryResult shardOperationOnPrimary(BulkShardRequest request, IndexShard primary) throws Exception {
+    public WritePrimaryResult<BulkShardRequest, BulkShardResponse> shardOperationOnPrimary(
+            BulkShardRequest request, IndexShard primary) throws Exception {
         final IndexMetaData metaData = primary.indexSettings().getIndexMetaData();
 
         long[] preVersions = new long[request.items().length];
@@ -119,7 +120,7 @@ protected WritePrimaryResult shardOperationOnPrimary(BulkShardRequest request, I
             responses[i] = items[i].getPrimaryResponse();
         }
         BulkShardResponse response = new BulkShardResponse(request.shardId(), responses);
-        return new WritePrimaryResult(request, response, location, null, primary);
+        return new WritePrimaryResult<>(request, response, location, null, primary, logger);
     }
 
     /** Executes bulk item requests and handles request execution exceptions */
@@ -326,7 +327,7 @@ private UpdateResultHolder executeUpdateRequest(UpdateRequest updateRequest, Ind
     }
 
     @Override
-    protected WriteReplicaResult shardOperationOnReplica(BulkShardRequest request, IndexShard replica) throws Exception {
+    public WriteReplicaResult<BulkShardRequest> shardOperationOnReplica(BulkShardRequest request, IndexShard replica) throws Exception {
         Translog.Location location = null;
         for (int i = 0; i < request.items().length; i++) {
             BulkItemRequest item = request.items()[i];
@@ -367,7 +368,89 @@ protected WriteReplicaResult shardOperationOnReplica(BulkShardRequest request, I
                 }
             }
         }
-        return new WriteReplicaResult(request, location, null, replica);
+        return new WriteReplicaResult<>(request, location, null, replica, logger);
+    }
+
+    /**
+     * Execute the given {@link IndexRequest} on a replica shard, throwing a
+     * {@link RetryOnReplicaException} if the operation needs to be re-tried.
+     */
+    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;
+        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);
+        }
+        return replica.index(operation);
+    }
+
+    /** Utility method to prepare an index operation on primary shards */
+    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 primary.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
+    }
+
+    /** Executes index operation on primary shard after updates mapping if dynamic mappings are found */
+    public static Engine.IndexResult executeIndexRequestOnPrimary(IndexRequest request, IndexShard primary,
+                                                                  MappingUpdatedAction mappingUpdatedAction) throws Exception {
+        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 = primary.shardId();
+        if (update != null) {
+            // 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");
+            }
+        }
+        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;
+    }
+
+    public static Engine.DeleteResult executeDeleteRequestOnPrimary(DeleteRequest request, IndexShard primary) {
+        final Engine.Delete delete = primary.prepareDeleteOnPrimary(request.type(), request.id(), request.version(), request.versionType());
+        return primary.delete(delete);
+    }
+
+    public static Engine.DeleteResult executeDeleteRequestOnReplica(DeleteRequest request, IndexShard replica) {
+        final Engine.Delete delete = replica.prepareDeleteOnReplica(request.type(), request.id(),
+                request.version(), request.versionType());
+        return replica.delete(delete);
     }
 
     private Translog.Location locationToSync(Translog.Location current, Translog.Location next) {

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

@@ -0,0 +1,132 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.bulk;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.support.WriteResponse;
+import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
+import org.elasticsearch.action.support.replication.TransportWriteAction;
+import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.function.Supplier;
+
+/** use transport bulk action directly */
+@Deprecated
+public abstract class TransportSingleItemBulkWriteAction<
+    Request extends ReplicatedWriteRequest<Request>,
+    Response extends ReplicationResponse & WriteResponse
+    > extends TransportWriteAction<Request, Request, Response> {
+
+    private final TransportBulkAction bulkAction;
+    private final TransportShardBulkAction shardBulkAction;
+
+
+    protected TransportSingleItemBulkWriteAction(Settings settings, String actionName, TransportService transportService,
+                                                 ClusterService clusterService, IndicesService indicesService, ThreadPool threadPool,
+                                                 ShardStateAction shardStateAction, ActionFilters actionFilters,
+                                                 IndexNameExpressionResolver indexNameExpressionResolver, Supplier<Request> request,
+                                                 Supplier<Request> replicaRequest, String executor,
+                                                 TransportBulkAction bulkAction, TransportShardBulkAction shardBulkAction) {
+        super(settings, actionName, transportService, clusterService, indicesService, threadPool, shardStateAction, actionFilters,
+            indexNameExpressionResolver, request, replicaRequest, executor);
+        this.bulkAction = bulkAction;
+        this.shardBulkAction = shardBulkAction;
+    }
+
+
+    @Override
+    protected void doExecute(Task task, final Request request, final ActionListener<Response> listener) {
+        bulkAction.execute(task, toSingleItemBulkRequest(request), wrapBulkResponse(listener));
+    }
+
+    @Override
+    protected WritePrimaryResult<Request, Response> shardOperationOnPrimary(
+        Request request, final IndexShard primary) throws Exception {
+        BulkItemRequest[] itemRequests = new BulkItemRequest[1];
+        WriteRequest.RefreshPolicy refreshPolicy = request.getRefreshPolicy();
+        request.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE);
+        itemRequests[0] = new BulkItemRequest(0, ((DocWriteRequest) request));
+        BulkShardRequest bulkShardRequest = new BulkShardRequest(request.shardId(), refreshPolicy, itemRequests);
+        WritePrimaryResult<BulkShardRequest, BulkShardResponse> bulkResult =
+            shardBulkAction.shardOperationOnPrimary(bulkShardRequest, primary);
+        assert bulkResult.finalResponseIfSuccessful.getResponses().length == 1 : "expected only one bulk shard response";
+        BulkItemResponse itemResponse = bulkResult.finalResponseIfSuccessful.getResponses()[0];
+        final Response response;
+        final Exception failure;
+        if (itemResponse.isFailed()) {
+            failure = itemResponse.getFailure().getCause();
+            response = null;
+        } else {
+            response = (Response) itemResponse.getResponse();
+            failure = null;
+        }
+        return new WritePrimaryResult<>(request, response, bulkResult.location, failure, primary, logger);
+    }
+
+    @Override
+    protected WriteReplicaResult<Request> shardOperationOnReplica(
+        Request replicaRequest, IndexShard replica) throws Exception {
+        BulkItemRequest[] itemRequests = new BulkItemRequest[1];
+        WriteRequest.RefreshPolicy refreshPolicy = replicaRequest.getRefreshPolicy();
+        itemRequests[0] = new BulkItemRequest(0, ((DocWriteRequest) replicaRequest));
+        BulkShardRequest bulkShardRequest = new BulkShardRequest(replicaRequest.shardId(), refreshPolicy, itemRequests);
+        WriteReplicaResult<BulkShardRequest> result = shardBulkAction.shardOperationOnReplica(bulkShardRequest, replica);
+        // a replica operation can never throw a document-level failure,
+        // as the same document has been already indexed successfully in the primary
+        return new WriteReplicaResult<>(replicaRequest, result.location, null, replica, logger);
+    }
+
+
+    private ActionListener<BulkResponse> wrapBulkResponse(ActionListener<Response> listener) {
+        return ActionListener.wrap(bulkItemResponses -> {
+            assert bulkItemResponses.getItems().length == 1 : "expected only one item in bulk request";
+            BulkItemResponse bulkItemResponse = bulkItemResponses.getItems()[0];
+            if (bulkItemResponse.isFailed() == false) {
+                final DocWriteResponse response = bulkItemResponse.getResponse();
+                listener.onResponse((Response) response);
+            } else {
+                listener.onFailure(bulkItemResponse.getFailure().getCause());
+            }
+        }, listener::onFailure);
+    }
+
+    public static BulkRequest toSingleItemBulkRequest(ReplicatedWriteRequest request) {
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(((DocWriteRequest) request));
+        bulkRequest.setRefreshPolicy(request.getRefreshPolicy());
+        bulkRequest.timeout(request.timeout());
+        bulkRequest.waitForActiveShards(request.waitForActiveShards());
+        request.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE);
+        return bulkRequest;
+    }
+}

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.delete;
 
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.CompositeIndicesRequest;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
 import org.elasticsearch.common.Nullable;
@@ -45,7 +46,7 @@
  * @see org.elasticsearch.client.Client#delete(DeleteRequest)
  * @see org.elasticsearch.client.Requests#deleteRequest(String)
  */
-public class DeleteRequest extends ReplicatedWriteRequest<DeleteRequest> implements DocWriteRequest<DeleteRequest> {
+public class DeleteRequest extends ReplicatedWriteRequest<DeleteRequest> implements DocWriteRequest<DeleteRequest>, CompositeIndicesRequest {
 
     private String type;
     private String id;