Bulk api: fail deletes when routing is required but not specified

As part of elastic#10136 we removed the transport action for broadcast deletes in case routing is required but not specified. Bulk api worked differently though and kept on doing the broadcast delete internally in that case. This commit makes sure that delete items are marked as failed in such cases. Also the check has been moved up in the code together with the existing check for the update api, and we now make sure that the exception is the same as the one thrown for single document apis (delete/update).

Note that the failure for the update api contained the wrong optype (the type of the document rather than "update"), that's been fixed too and tested.

Closes elastic#16645

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

@@ -31,10 +31,12 @@
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
 import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.TransportDeleteAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.action.update.TransportUpdateAction;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
@@ -43,8 +45,6 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MappingMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.routing.GroupShardsIterator;
-import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
@@ -201,7 +201,7 @@ public void executeBulk(final BulkRequest bulkRequest, final ActionListener<Bulk
         executeBulk(bulkRequest, startTime, listener, new AtomicArray<BulkItemResponse>(bulkRequest.requests.size()));
     }
 
-    private final long buildTookInMillis(long startTime) {
+    private long buildTookInMillis(long startTime) {
         // protect ourselves against time going backwards
         return Math.max(1, System.currentTimeMillis() - startTime);
     }
@@ -215,33 +215,52 @@ private void executeBulk(final BulkRequest bulkRequest, final long startTime, fi
         MetaData metaData = clusterState.metaData();
         for (int i = 0; i < bulkRequest.requests.size(); i++) {
             ActionRequest request = bulkRequest.requests.get(i);
-            if (request instanceof DocumentRequest) {
-                DocumentRequest req = (DocumentRequest) request;
-
-                if (addFailureIfIndexIsUnavailable(req, bulkRequest, responses, i, concreteIndices, metaData)) {
-                    continue;
+            //the request can only be null because we set it to null in the previous step, so it gets ignored
+            if (request == null) {
+                continue;
+            }
+            DocumentRequest documentRequest = (DocumentRequest) request;
+            if (addFailureIfIndexIsUnavailable(documentRequest, bulkRequest, responses, i, concreteIndices, metaData)) {
+                continue;
+            }
+            String concreteIndex = concreteIndices.resolveIfAbsent(documentRequest);
+            if (request instanceof IndexRequest) {
+                IndexRequest indexRequest = (IndexRequest) request;
+                MappingMetaData mappingMd = null;
+                if (metaData.hasIndex(concreteIndex)) {
+                    mappingMd = metaData.index(concreteIndex).mappingOrDefault(indexRequest.type());
                 }
-
-                String concreteIndex = concreteIndices.resolveIfAbsent(req);
-                if (request instanceof IndexRequest) {
-                    IndexRequest indexRequest = (IndexRequest) request;
-                    MappingMetaData mappingMd = null;
-                    if (metaData.hasIndex(concreteIndex)) {
-                        mappingMd = metaData.index(concreteIndex).mappingOrDefault(indexRequest.type());
-                    }
-                    try {
-                        indexRequest.process(metaData, mappingMd, allowIdGeneration, concreteIndex);
-                    } catch (ElasticsearchParseException | RoutingMissingException e) {
-                        BulkItemResponse.Failure failure = new BulkItemResponse.Failure(concreteIndex, indexRequest.type(), indexRequest.id(), e);
-                        BulkItemResponse bulkItemResponse = new BulkItemResponse(i, "index", failure);
-                        responses.set(i, bulkItemResponse);
-                        // make sure the request gets never processed again
-                        bulkRequest.requests.set(i, null);
-                    }
-                } else {
-                    concreteIndices.resolveIfAbsent(req);
-                    req.routing(clusterState.metaData().resolveIndexRouting(req.routing(), req.index()));
+                try {
+                    indexRequest.process(metaData, mappingMd, allowIdGeneration, concreteIndex);
+                } catch (ElasticsearchParseException | RoutingMissingException e) {
+                    BulkItemResponse.Failure failure = new BulkItemResponse.Failure(concreteIndex, indexRequest.type(), indexRequest.id(), e);
+                    BulkItemResponse bulkItemResponse = new BulkItemResponse(i, "index", failure);
+                    responses.set(i, bulkItemResponse);
+                    // make sure the request gets never processed again
+                    bulkRequest.requests.set(i, null);
                 }
+            } else if (request instanceof DeleteRequest) {
+                try {
+                    TransportDeleteAction.resolveAndValidateRouting(metaData, concreteIndex, (DeleteRequest)request);
+                } catch(RoutingMissingException e) {
+                    BulkItemResponse.Failure failure = new BulkItemResponse.Failure(concreteIndex, documentRequest.type(), documentRequest.id(), e);
+                    BulkItemResponse bulkItemResponse = new BulkItemResponse(i, "delete", failure);
+                    responses.set(i, bulkItemResponse);
+                    // make sure the request gets never processed again
+                    bulkRequest.requests.set(i, null);
+                }
+            } else if (request instanceof UpdateRequest) {
+                try {
+                    TransportUpdateAction.resolveAndValidateRouting(metaData, concreteIndex, (UpdateRequest)request);
+                } catch(RoutingMissingException e) {
+                    BulkItemResponse.Failure failure = new BulkItemResponse.Failure(concreteIndex, documentRequest.type(), documentRequest.id(), e);
+                    BulkItemResponse bulkItemResponse = new BulkItemResponse(i, "update", failure);
+                    responses.set(i, bulkItemResponse);
+                    // make sure the request gets never processed again
+                    bulkRequest.requests.set(i, null);
+                }
+            } else {
+                throw new AssertionError("request type not supported: [" + request.getClass().getName() + "]");
             }
         }
 
@@ -263,37 +282,16 @@ private void executeBulk(final BulkRequest bulkRequest, final long startTime, fi
             } else if (request instanceof DeleteRequest) {
                 DeleteRequest deleteRequest = (DeleteRequest) request;
                 String concreteIndex = concreteIndices.getConcreteIndex(deleteRequest.index());
-                MappingMetaData mappingMd = clusterState.metaData().index(concreteIndex).mappingOrDefault(deleteRequest.type());
-                if (mappingMd != null && mappingMd.routing().required() && deleteRequest.routing() == null) {
-                    // if routing is required, and no routing on the delete request, we need to broadcast it....
-                    GroupShardsIterator groupShards = clusterService.operationRouting().broadcastDeleteShards(clusterState, concreteIndex);
-                    for (ShardIterator shardIt : groupShards) {
-                        List<BulkItemRequest> list = requestsByShard.get(shardIt.shardId());
-                        if (list == null) {
-                            list = new ArrayList<>();
-                            requestsByShard.put(shardIt.shardId(), list);
-                        }
-                        list.add(new BulkItemRequest(i, new DeleteRequest(deleteRequest)));
-                    }
-                } else {
-                    ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex, deleteRequest.type(), deleteRequest.id(), deleteRequest.routing()).shardId();
-                    List<BulkItemRequest> list = requestsByShard.get(shardId);
-                    if (list == null) {
-                        list = new ArrayList<>();
-                        requestsByShard.put(shardId, list);
-                    }
-                    list.add(new BulkItemRequest(i, request));
+                ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex, deleteRequest.type(), deleteRequest.id(), deleteRequest.routing()).shardId();
+                List<BulkItemRequest> list = requestsByShard.get(shardId);
+                if (list == null) {
+                    list = new ArrayList<>();
+                    requestsByShard.put(shardId, list);
                 }
+                list.add(new BulkItemRequest(i, request));
             } else if (request instanceof UpdateRequest) {
                 UpdateRequest updateRequest = (UpdateRequest) request;
                 String concreteIndex = concreteIndices.getConcreteIndex(updateRequest.index());
-                MappingMetaData mappingMd = clusterState.metaData().index(concreteIndex).mappingOrDefault(updateRequest.type());
-                if (mappingMd != null && mappingMd.routing().required() && updateRequest.routing() == null) {
-                    BulkItemResponse.Failure failure = new BulkItemResponse.Failure(updateRequest.index(), updateRequest.type(),
-                            updateRequest.id(), new IllegalArgumentException("routing is required for this item"));
-                    responses.set(i, new BulkItemResponse(i, updateRequest.type(), failure));
-                    continue;
-                }
                 ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex, updateRequest.type(), updateRequest.id(), updateRequest.routing()).shardId();
                 List<BulkItemRequest> list = requestsByShard.get(shardId);
                 if (list == null) {

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

@@ -95,23 +95,28 @@ public void onFailure(Throwable e) {
 
     @Override
     protected void resolveRequest(final MetaData metaData, String concreteIndex, DeleteRequest request) {
+        resolveAndValidateRouting(metaData, concreteIndex, request);
+        ShardId shardId = clusterService.operationRouting().shardId(clusterService.state(), concreteIndex, request.type(),
+                request.id(), request.routing());
+        request.setShardId(shardId);
+    }
+
+    public static void resolveAndValidateRouting(final MetaData metaData, String concreteIndex, DeleteRequest request) {
         request.routing(metaData.resolveIndexRouting(request.routing(), request.index()));
         if (metaData.hasIndex(concreteIndex)) {
-            // check if routing is required, if so, do a broadcast delete
+            // check if routing is required, if so, throw error if routing wasn't specified
             MappingMetaData mappingMd = metaData.index(concreteIndex).mappingOrDefault(request.type());
             if (mappingMd != null && mappingMd.routing().required()) {
                 if (request.routing() == null) {
                     if (request.versionType() != VersionType.INTERNAL) {
                         // TODO: implement this feature
                         throw new IllegalArgumentException("routing value is required for deleting documents of type [" + request.type()
-                                + "] while using version_type [" + request.versionType() + "]");
+                            + "] while using version_type [" + request.versionType() + "]");
                     }
                     throw new RoutingMissingException(concreteIndex, request.type(), request.id());
                 }
             }
         }
-        ShardId shardId = clusterService.operationRouting().shardId(clusterService.state(), concreteIndex, request.type(), request.id(), request.routing());
-        request.setShardId(shardId);
     }
 
     private void innerExecute(final DeleteRequest request, final ActionListener<DeleteResponse> listener) {

core/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.routing.PlainShardIterator;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -102,12 +103,16 @@ protected boolean retryOnFailure(Throwable e) {
 
     @Override
     protected boolean resolveRequest(ClusterState state, UpdateRequest request, ActionListener<UpdateResponse> listener) {
-        request.routing((state.metaData().resolveIndexRouting(request.routing(), request.index())));
+        resolveAndValidateRouting(state.metaData(), request.concreteIndex(), request);
+        return true;
+    }
+
+    public static void resolveAndValidateRouting(MetaData metaData, String concreteIndex, UpdateRequest request) {
+        request.routing((metaData.resolveIndexRouting(request.routing(), request.index())));
         // Fail fast on the node that received the request, rather than failing when translating on the index or delete request.
-        if (request.routing() == null && state.getMetaData().routingRequired(request.concreteIndex(), request.type())) {
-            throw new RoutingMissingException(request.concreteIndex(), request.type(), request.id());
+        if (request.routing() == null && metaData.routingRequired(concreteIndex, request.type())) {
+            throw new RoutingMissingException(concreteIndex, request.type(), request.id());
         }
-        return true;
     }
 
     @Override

core/src/main/java/org/elasticsearch/cluster/routing/OperationRouting.java

@@ -69,10 +69,6 @@ public ShardIterator getShards(ClusterState clusterState, String index, int shar
         return preferenceActiveShardIterator(indexShard, clusterState.nodes().localNodeId(), clusterState.nodes(), preference);
     }
 
-    public GroupShardsIterator broadcastDeleteShards(ClusterState clusterState, String index) {
-        return indexRoutingTable(clusterState, index).groupByShardsIt();
-    }
-
     public int searchShardsCount(ClusterState clusterState, String[] concreteIndices, @Nullable Map<String, Set<String>> routing) {
         final Set<IndexShardRoutingTable> shards = computeTargetedShards(clusterState, concreteIndices, routing);
         return shards.size();