Remove custom execute local logic in TransportSingleShardAction and TransportInstanceSingleOperationAction and rely on transport service to execute locally. (forking thread etc.)

Change TransportInstanceSingleOperationAction to have shardActionHandler to, so we can execute locally without endless spinning.

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

@@ -126,12 +126,10 @@ public class BulkRequest extends ActionRequest<BulkRequest> implements Composite
      * Adds an {@link UpdateRequest} to the list of actions to execute.
      */
     public BulkRequest add(UpdateRequest request) {
-        request.beforeLocalFork();
         return internalAdd(request, null);
     }
 
     public BulkRequest add(UpdateRequest request, @Nullable Object payload) {
-        request.beforeLocalFork();
         return internalAdd(request, payload);
     }
 

core/src/main/java/org/elasticsearch/action/support/single/instance/InstanceShardOperationRequest.java

@@ -44,6 +44,8 @@ public abstract class InstanceShardOperationRequest<T extends InstanceShardOpera
     // -1 means its not set, allows to explicitly direct a request to a specific shard
     protected int shardId = -1;
 
+    private String concreteIndex;
+
     protected InstanceShardOperationRequest() {
     }
 
@@ -100,13 +102,21 @@ public abstract class InstanceShardOperationRequest<T extends InstanceShardOpera
         return timeout(TimeValue.parseTimeValue(timeout, null, getClass().getSimpleName() + ".timeout"));
     }
 
+    public String concreteIndex() {
+        return concreteIndex;
+    }
+
+    void concreteIndex(String concreteIndex) {
+        this.concreteIndex = concreteIndex;
+    }
+
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         index = in.readString();
         shardId = in.readInt();
         timeout = TimeValue.readTimeValue(in);
-        // no need to pass threading over the network, they are always false when coming throw a thread pool
+        concreteIndex = in.readOptionalString();
     }
 
     @Override
@@ -115,9 +125,8 @@ public abstract class InstanceShardOperationRequest<T extends InstanceShardOpera
         out.writeString(index);
         out.writeInt(shardId);
         timeout.writeTo(out);
+        out.writeOptionalString(concreteIndex);
     }
 
-    public void beforeLocalFork() {
-    }
 }
 

core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java

@@ -53,6 +53,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
     protected final TransportService transportService;
 
     final String executor;
+    final String shardActionName;
 
     protected TransportInstanceSingleOperationAction(Settings settings, String actionName, ThreadPool threadPool,
                                                      ClusterService clusterService, TransportService transportService,
@@ -61,6 +62,8 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
         this.clusterService = clusterService;
         this.transportService = transportService;
         this.executor = executor();
+        this.shardActionName = actionName + "[s]";
+        transportService.registerRequestHandler(shardActionName, request, executor, new ShardTransportHandler());
     }
 
     @Override
@@ -70,7 +73,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
 
     protected abstract String executor();
 
-    protected abstract void shardOperation(InternalRequest request, ActionListener<Response> listener);
+    protected abstract void shardOperation(Request request, ActionListener<Response> listener);
 
     protected abstract Response newResponse();
 
@@ -78,14 +81,14 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
         return state.blocks().globalBlockedException(ClusterBlockLevel.WRITE);
     }
 
-    protected ClusterBlockException checkRequestBlock(ClusterState state, InternalRequest request) {
+    protected ClusterBlockException checkRequestBlock(ClusterState state, Request request) {
         return state.blocks().indexBlockedException(ClusterBlockLevel.WRITE, request.concreteIndex());
     }
     /**
      * Resolves the request. If the resolve means a different execution, then return false
      * here to indicate not to continue and execute this request.
      */
-    protected abstract boolean resolveRequest(ClusterState state, InternalRequest request, ActionListener<Response> listener);
+    protected abstract boolean resolveRequest(ClusterState state, Request request, ActionListener<Response> listener);
 
     protected boolean retryOnFailure(Throwable e) {
         return false;
@@ -98,24 +101,24 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
     /**
      * Should return an iterator with a single shard!
      */
-    protected abstract ShardIterator shards(ClusterState clusterState, InternalRequest request);
+    protected abstract ShardIterator shards(ClusterState clusterState, Request request);
 
     class AsyncSingleAction {
 
         private final ActionListener<Response> listener;
-        private final InternalRequest internalRequest;
+        private final Request request;
         private volatile ClusterStateObserver observer;
         private ShardIterator shardIt;
         private DiscoveryNodes nodes;
         private final AtomicBoolean operationStarted = new AtomicBoolean();
 
         private AsyncSingleAction(Request request, ActionListener<Response> listener) {
-            this.internalRequest = new InternalRequest(request);
+            this.request = request;
             this.listener = listener;
         }
 
         public void start() {
-            this.observer = new ClusterStateObserver(clusterService, internalRequest.request().timeout(), logger);
+            this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger);
             doStart();
         }
 
@@ -131,12 +134,12 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                         throw blockException;
                     }
                 }
-                internalRequest.concreteIndex(indexNameExpressionResolver.concreteSingleIndex(observer.observedState(), internalRequest.request()));
+                request.concreteIndex(indexNameExpressionResolver.concreteSingleIndex(observer.observedState(), request));
                 // check if we need to execute, and if not, return
-                if (!resolveRequest(observer.observedState(), internalRequest, listener)) {
+                if (!resolveRequest(observer.observedState(), request, listener)) {
                     return true;
                 }
-                blockException = checkRequestBlock(observer.observedState(), internalRequest);
+                blockException = checkRequestBlock(observer.observedState(), request);
                 if (blockException != null) {
                     if (blockException.retryable()) {
                         retry(blockException);
@@ -145,7 +148,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                         throw blockException;
                     }
                 }
-                shardIt = shards(observer.observedState(), internalRequest);
+                shardIt = shards(observer.observedState(), request);
             } catch (Throwable e) {
                 listener.onFailure(e);
                 return true;
@@ -172,68 +175,39 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                 return true;
             }
 
-            internalRequest.request().shardId = shardIt.shardId().id();
-            if (shard.currentNodeId().equals(nodes.localNodeId())) {
-                internalRequest.request().beforeLocalFork();
-                try {
-                    threadPool.executor(executor).execute(new Runnable() {
-                        @Override
-                        public void run() {
-                            try {
-                                shardOperation(internalRequest, listener);
-                            } catch (Throwable e) {
-                                if (retryOnFailure(e)) {
-                                    operationStarted.set(false);
-                                    // we already marked it as started when we executed it (removed the listener) so pass false
-                                    // to re-add to the cluster listener
-                                    retry(null);
-                                } else {
-                                    listener.onFailure(e);
-                                }
-                            }
-                        }
-                    });
-                } catch (Throwable e) {
-                    if (retryOnFailure(e)) {
-                        retry(null);
-                    } else {
-                        listener.onFailure(e);
-                    }
-                }
-            } else {
-                DiscoveryNode node = nodes.get(shard.currentNodeId());
-                transportService.sendRequest(node, actionName, internalRequest.request(), transportOptions(), new BaseTransportResponseHandler<Response>() {
+            request.shardId = shardIt.shardId().id();
+            DiscoveryNode node = nodes.get(shard.currentNodeId());
+            transportService.sendRequest(node, shardActionName, request, transportOptions(), new BaseTransportResponseHandler<Response>() {
 
-                    @Override
-                    public Response newInstance() {
-                        return newResponse();
-                    }
+                @Override
+                public Response newInstance() {
+                    return newResponse();
+                }
 
-                    @Override
-                    public String executor() {
-                        return ThreadPool.Names.SAME;
-                    }
+                @Override
+                public String executor() {
+                    return ThreadPool.Names.SAME;
+                }
 
-                    @Override
-                    public void handleResponse(Response response) {
-                        listener.onResponse(response);
-                    }
+                @Override
+                public void handleResponse(Response response) {
+                    listener.onResponse(response);
+                }
 
-                    @Override
-                    public void handleException(TransportException exp) {
-                        // if we got disconnected from the node, or the node / shard is not in the right state (being closed)
-                        if (exp.unwrapCause() instanceof ConnectTransportException || exp.unwrapCause() instanceof NodeClosedException ||
-                                retryOnFailure(exp)) {
-                            operationStarted.set(false);
-                            // we already marked it as started when we executed it (removed the listener) so pass false
-                            // to re-add to the cluster listener
-                            retry(null);
-                        } else {
-                            listener.onFailure(exp);
-                        }
+                @Override
+                public void handleException(TransportException exp) {
+                    // if we got disconnected from the node, or the node / shard is not in the right state (being closed)
+                    if (exp.unwrapCause() instanceof ConnectTransportException || exp.unwrapCause() instanceof NodeClosedException ||
+                            retryOnFailure(exp)) {
+                        operationStarted.set(false);
+                        // we already marked it as started when we executed it (removed the listener) so pass false
+                        // to re-add to the cluster listener
+                        retry(null);
+                    } else {
+                        listener.onFailure(exp);
                     }
-                });
-            }
+                }
+            });
             return true;
         }
 
@@ -243,8 +217,6 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                 return;
             }
 
-            // make it threaded operation so we fork on the discovery listener thread
-            internalRequest.request().beforeLocalFork();
             observer.waitForNextChange(new ClusterStateObserver.Listener() {
                 @Override
                 public void onNewClusterState(ClusterState state) {
@@ -263,39 +235,42 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                         Throwable listenFailure = failure;
                         if (listenFailure == null) {
                             if (shardIt == null) {
-                                listenFailure = new UnavailableShardsException(new ShardId(internalRequest.concreteIndex(), -1), "Timeout waiting for [" + timeout + "], request: " + internalRequest.request().toString());
+                                listenFailure = new UnavailableShardsException(new ShardId(request.concreteIndex(), -1), "Timeout waiting for [" + timeout + "], request: " + request.toString());
                             } else {
-                                listenFailure = new UnavailableShardsException(shardIt.shardId(), "[" + shardIt.size() + "] shardIt, [" + shardIt.sizeActive() + "] active : Timeout waiting for [" + timeout + "], request: " + internalRequest.request().toString());
+                                listenFailure = new UnavailableShardsException(shardIt.shardId(), "[" + shardIt.size() + "] shardIt, [" + shardIt.sizeActive() + "] active : Timeout waiting for [" + timeout + "], request: " + request.toString());
                             }
                         }
                         listener.onFailure(listenFailure);
                     }
                 }
-            }, internalRequest.request().timeout());
+            }, request.timeout());
         }
     }
 
-    /**
-     * Internal request class that gets built on each node. Holds the original request plus additional info.
-     */
-    protected class InternalRequest {
-        final Request request;
-        String concreteIndex;
-
-        InternalRequest(Request request) {
-            this.request = request;
-        }
+    private class ShardTransportHandler implements TransportRequestHandler<Request> {
 
-        public Request request() {
-            return request;
-        }
+        @Override
+        public void messageReceived(final Request request, final TransportChannel channel) throws Exception {
+            shardOperation(request, new ActionListener<Response>() {
+                @Override
+                public void onResponse(Response response) {
+                    try {
+                        channel.sendResponse(response);
+                    } catch (Throwable e) {
+                        onFailure(e);
+                    }
+                }
 
-        void concreteIndex(String concreteIndex) {
-            this.concreteIndex = concreteIndex;
-        }
+                @Override
+                public void onFailure(Throwable e) {
+                    try {
+                        channel.sendResponse(e);
+                    } catch (Exception e1) {
+                        logger.warn("failed to send response for get", e1);
+                    }
+                }
+            });
 
-        public String concreteIndex() {
-            return concreteIndex;
         }
     }
 }

core/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java

@@ -172,59 +172,33 @@ public abstract class TransportSingleShardAction<Request extends SingleShardRequ
                 listener.onFailure(failure);
                 return;
             }
-            if (shardRouting.currentNodeId().equals(nodes.localNodeId())) {
-                if (logger.isTraceEnabled()) {
-                    logger.trace("executing [{}] on shard [{}]", internalRequest.request(), shardRouting.shardId());
-                }
-                try {
-                    if (internalRequest.request().operationThreaded()) {
-                        threadPool.executor(executor).execute(new Runnable() {
-                            @Override
-                            public void run() {
-                                try {
-                                    Response response = shardOperation(internalRequest.request(), shardRouting.shardId());
-                                    listener.onResponse(response);
-                                } catch (Throwable e) {
-                                    onFailure(shardRouting, e);
-                                }
-                            }
-                        });
-                    } else {
-                        final Response response = shardOperation(internalRequest.request(), shardRouting.shardId());
+            DiscoveryNode node = nodes.get(shardRouting.currentNodeId());
+            if (node == null) {
+                onFailure(shardRouting, new NoShardAvailableActionException(shardIt.shardId()));
+            } else {
+                internalRequest.request().internalShardId = shardRouting.shardId();
+                transportService.sendRequest(node, transportShardAction, internalRequest.request(), new BaseTransportResponseHandler<Response>() {
+
+                    @Override
+                    public Response newInstance() {
+                        return newResponse();
+                    }
+
+                    @Override
+                    public String executor() {
+                        return ThreadPool.Names.SAME;
+                    }
+
+                    @Override
+                    public void handleResponse(final Response response) {
                         listener.onResponse(response);
                     }
-                } catch (Throwable e) {
-                    onFailure(shardRouting, e);
-                }
-            } else {
-                DiscoveryNode node = nodes.get(shardRouting.currentNodeId());
-                if (node == null) {
-                    onFailure(shardRouting, new NoShardAvailableActionException(shardIt.shardId()));
-                } else {
-                    internalRequest.request().internalShardId = shardRouting.shardId();
-                    transportService.sendRequest(node, transportShardAction, internalRequest.request(), new BaseTransportResponseHandler<Response>() {
-
-                        @Override
-                        public Response newInstance() {
-                            return newResponse();
-                        }
-
-                        @Override
-                        public String executor() {
-                            return ThreadPool.Names.SAME;
-                        }
-
-                        @Override
-                        public void handleResponse(final Response response) {
-                            listener.onResponse(response);
-                        }
-
-                        @Override
-                        public void handleException(TransportException exp) {
-                            onFailure(shardRouting, exp);
-                        }
-                    });
-                }
+
+                    @Override
+                    public void handleException(TransportException exp) {
+                        onFailure(shardRouting, exp);
+                    }
+                });
             }
         }
     }

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

@@ -101,11 +101,11 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
     }
 
     @Override
-    protected boolean resolveRequest(ClusterState state, InternalRequest request, ActionListener<UpdateResponse> listener) {
-        request.request().routing((state.metaData().resolveIndexRouting(request.request().routing(), request.request().index())));
+    protected boolean resolveRequest(ClusterState state, UpdateRequest request, ActionListener<UpdateResponse> listener) {
+        request.routing((state.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.request().routing() == null && state.getMetaData().routingRequired(request.concreteIndex(), request.request().type())) {
-            throw new RoutingMissingException(request.concreteIndex(), request.request().type(), request.request().id());
+        if (request.routing() == null && state.getMetaData().routingRequired(request.concreteIndex(), request.type())) {
+            throw new RoutingMissingException(request.concreteIndex(), request.type(), request.id());
         }
         return true;
     }
@@ -114,7 +114,6 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
     protected void doExecute(final UpdateRequest request, final ActionListener<UpdateResponse> listener) {
         // if we don't have a master, we don't have metadata, that's fine, let it find a master using create index API
         if (autoCreateIndex.shouldAutoCreate(request.index(), clusterService.state())) {
-            request.beforeLocalFork(); // we fork on another thread...
             createIndexAction.execute(new CreateIndexRequest(request).index(request.index()).cause("auto(update api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() {
                 @Override
                 public void onResponse(CreateIndexResponse result) {
@@ -145,12 +144,12 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
     }
 
     @Override
-    protected ShardIterator shards(ClusterState clusterState, InternalRequest request) {
-        if (request.request().shardId() != -1) {
-            return clusterState.routingTable().index(request.concreteIndex()).shard(request.request().shardId()).primaryShardIt();
+    protected ShardIterator shards(ClusterState clusterState, UpdateRequest request) {
+        if (request.shardId() != -1) {
+            return clusterState.routingTable().index(request.concreteIndex()).shard(request.shardId()).primaryShardIt();
         }
         ShardIterator shardIterator = clusterService.operationRouting()
-                .indexShards(clusterState, request.concreteIndex(), request.request().type(), request.request().id(), request.request().routing());
+                .indexShards(clusterState, request.concreteIndex(), request.type(), request.id(), request.routing());
         ShardRouting shard;
         while ((shard = shardIterator.nextOrNull()) != null) {
             if (shard.primary()) {
@@ -161,26 +160,26 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
     }
 
     @Override
-    protected void shardOperation(final InternalRequest request, final ActionListener<UpdateResponse> listener) {
+    protected void shardOperation(final UpdateRequest request, final ActionListener<UpdateResponse> listener) {
         shardOperation(request, listener, 0);
     }
 
-    protected void shardOperation(final InternalRequest request, final ActionListener<UpdateResponse> listener, final int retryCount) {
+    protected void shardOperation(final UpdateRequest request, final ActionListener<UpdateResponse> listener, final int retryCount) {
         IndexService indexService = indicesService.indexServiceSafe(request.concreteIndex());
-        IndexShard indexShard = indexService.shardSafe(request.request().shardId());
-        final UpdateHelper.Result result = updateHelper.prepare(request.request(), indexShard);
+        IndexShard indexShard = indexService.shardSafe(request.shardId());
+        final UpdateHelper.Result result = updateHelper.prepare(request, indexShard);
         switch (result.operation()) {
             case UPSERT:
-                IndexRequest upsertRequest = new IndexRequest((IndexRequest)result.action(), request.request());
+                IndexRequest upsertRequest = new IndexRequest((IndexRequest)result.action(), request);
                 // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request
                 final BytesReference upsertSourceBytes = upsertRequest.source();
                 indexAction.execute(upsertRequest, new ActionListener<IndexResponse>() {
                     @Override
                     public void onResponse(IndexResponse response) {
                         UpdateResponse update = new UpdateResponse(response.getShardInfo(), response.getIndex(), response.getType(), response.getId(), response.getVersion(), response.isCreated());
-                        if (request.request().fields() != null && request.request().fields().length > 0) {
+                        if (request.fields() != null && request.fields().length > 0) {
                             Tuple<XContentType, Map<String, Object>> sourceAndContent = XContentHelper.convertToMap(upsertSourceBytes, true);
-                            update.setGetResult(updateHelper.extractGetResult(request.request(), request.concreteIndex(), response.getVersion(), sourceAndContent.v2(), sourceAndContent.v1(), upsertSourceBytes));
+                            update.setGetResult(updateHelper.extractGetResult(request, request.concreteIndex(), response.getVersion(), sourceAndContent.v2(), sourceAndContent.v1(), upsertSourceBytes));
                         } else {
                             update.setGetResult(null);
                         }
@@ -191,7 +190,7 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                     public void onFailure(Throwable e) {
                         e = ExceptionsHelper.unwrapCause(e);
                         if (e instanceof VersionConflictEngineException || e instanceof DocumentAlreadyExistsException) {
-                            if (retryCount < request.request().retryOnConflict()) {
+                            if (retryCount < request.retryOnConflict()) {
                                 threadPool.executor(executor()).execute(new ActionRunnable<UpdateResponse>(listener) {
                                     @Override
                                     protected void doRun() {
@@ -206,14 +205,14 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                 });
                 break;
             case INDEX:
-                IndexRequest indexRequest = new IndexRequest((IndexRequest)result.action(), request.request());
+                IndexRequest indexRequest = new IndexRequest((IndexRequest)result.action(), request);
                 // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request
                 final BytesReference indexSourceBytes = indexRequest.source();
                 indexAction.execute(indexRequest, new ActionListener<IndexResponse>() {
                     @Override
                     public void onResponse(IndexResponse response) {
                         UpdateResponse update = new UpdateResponse(response.getShardInfo(), response.getIndex(), response.getType(), response.getId(), response.getVersion(), response.isCreated());
-                        update.setGetResult(updateHelper.extractGetResult(request.request(), request.concreteIndex(), response.getVersion(), result.updatedSourceAsMap(), result.updateSourceContentType(), indexSourceBytes));
+                        update.setGetResult(updateHelper.extractGetResult(request, request.concreteIndex(), response.getVersion(), result.updatedSourceAsMap(), result.updateSourceContentType(), indexSourceBytes));
                         listener.onResponse(update);
                     }
 
@@ -221,7 +220,7 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                     public void onFailure(Throwable e) {
                         e = ExceptionsHelper.unwrapCause(e);
                         if (e instanceof VersionConflictEngineException) {
-                            if (retryCount < request.request().retryOnConflict()) {
+                            if (retryCount < request.retryOnConflict()) {
                                 threadPool.executor(executor()).execute(new ActionRunnable<UpdateResponse>(listener) {
                                     @Override
                                     protected void doRun() {
@@ -236,12 +235,12 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                 });
                 break;
             case DELETE:
-                DeleteRequest deleteRequest = new DeleteRequest((DeleteRequest)result.action(), request.request());
+                DeleteRequest deleteRequest = new DeleteRequest((DeleteRequest)result.action(), request);
                 deleteAction.execute(deleteRequest, new ActionListener<DeleteResponse>() {
                     @Override
                     public void onResponse(DeleteResponse response) {
                         UpdateResponse update = new UpdateResponse(response.getShardInfo(), response.getIndex(), response.getType(), response.getId(), response.getVersion(), false);
-                        update.setGetResult(updateHelper.extractGetResult(request.request(), request.concreteIndex(), response.getVersion(), result.updatedSourceAsMap(), result.updateSourceContentType(), null));
+                        update.setGetResult(updateHelper.extractGetResult(request, request.concreteIndex(), response.getVersion(), result.updatedSourceAsMap(), result.updateSourceContentType(), null));
                         listener.onResponse(update);
                     }
 
@@ -249,7 +248,7 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                     public void onFailure(Throwable e) {
                         e = ExceptionsHelper.unwrapCause(e);
                         if (e instanceof VersionConflictEngineException) {
-                            if (retryCount < request.request().retryOnConflict()) {
+                            if (retryCount < request.retryOnConflict()) {
                                 threadPool.executor(executor()).execute(new ActionRunnable<UpdateResponse>(listener) {
                                     @Override
                                     protected void doRun() {
@@ -267,9 +266,9 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio
                 UpdateResponse update = result.action();
                 IndexService indexServiceOrNull = indicesService.indexService(request.concreteIndex());
                 if (indexServiceOrNull !=  null) {
-                    IndexShard shard = indexService.shard(request.request().shardId());
+                    IndexShard shard = indexService.shard(request.shardId());
                     if (shard != null) {
-                        shard.indexingService().noopUpdate(request.request().type());
+                        shard.indexingService().noopUpdate(request.type());
                     }
                 }
                 listener.onResponse(update);

core/src/test/java/org/elasticsearch/action/IndicesRequestTests.java

@@ -220,7 +220,7 @@ public class IndicesRequestTests extends ElasticsearchIntegrationTest {
     @Test
     public void testUpdate() {
         //update action goes to the primary, index op gets executed locally, then replicated
-        String[] updateShardActions = new String[]{UpdateAction.NAME, IndexAction.NAME + "[r]"};
+        String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", IndexAction.NAME + "[r]"};
         interceptTransportActions(updateShardActions);
 
         String indexOrAlias = randomIndexOrAlias();
@@ -236,7 +236,7 @@ public class IndicesRequestTests extends ElasticsearchIntegrationTest {
     @Test
     public void testUpdateUpsert() {
         //update action goes to the primary, index op gets executed locally, then replicated
-        String[] updateShardActions = new String[]{UpdateAction.NAME, IndexAction.NAME + "[r]"};
+        String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", IndexAction.NAME + "[r]"};
         interceptTransportActions(updateShardActions);
 
         String indexOrAlias = randomIndexOrAlias();
@@ -251,7 +251,7 @@ public class IndicesRequestTests extends ElasticsearchIntegrationTest {
     @Test
     public void testUpdateDelete() {
         //update action goes to the primary, delete op gets executed locally, then replicated
-        String[] updateShardActions = new String[]{UpdateAction.NAME, DeleteAction.NAME + "[r]"};
+        String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", DeleteAction.NAME + "[r]"};
         interceptTransportActions(updateShardActions);
 
         String indexOrAlias = randomIndexOrAlias();