diff --git a/core/src/main/java/org/elasticsearch/action/UnavailableShardsException.java b/core/src/main/java/org/elasticsearch/action/UnavailableShardsException.java
index dd0968ee4ea..ff31bb715db 100644
--- a/core/src/main/java/org/elasticsearch/action/UnavailableShardsException.java
+++ b/core/src/main/java/org/elasticsearch/action/UnavailableShardsException.java
@@ -21,6 +21,7 @@ package org.elasticsearch.action;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.collect.HppcMaps;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.rest.RestStatus;
@@ -32,8 +33,8 @@ import java.io.IOException;
  */
 public class UnavailableShardsException extends ElasticsearchException {
 
-    public UnavailableShardsException(@Nullable ShardId shardId, String message) {
-        super(buildMessage(shardId, message));
+    public UnavailableShardsException(@Nullable ShardId shardId, String message, Object... args) {
+        super(buildMessage(shardId, message), args);
     }
 
     private static String buildMessage(ShardId shardId, String message) {
diff --git a/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java b/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java
index 2e815da3835..5f4f9420295 100644
--- a/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java
+++ b/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.support.single.instance;
 
+import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.UnavailableShardsException;
@@ -35,6 +36,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.logging.support.LoggerMessageFormat;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.shard.ShardId;
@@ -42,6 +44,7 @@ import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
 
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
@@ -111,9 +114,8 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
         private volatile ClusterStateObserver observer;
         private ShardIterator shardIt;
         private DiscoveryNodes nodes;
-        private final AtomicBoolean operationStarted = new AtomicBoolean();
 
-        private AsyncSingleAction(Request request, ActionListener<Response> listener) {
+        AsyncSingleAction(Request request, ActionListener<Response> listener) {
             this.request = request;
             this.listener = listener;
         }
@@ -123,14 +125,14 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
             doStart();
         }
 
-        protected boolean doStart() {
+        protected void doStart() {
             nodes = observer.observedState().nodes();
             try {
                 ClusterBlockException blockException = checkGlobalBlock(observer.observedState());
                 if (blockException != null) {
                     if (blockException.retryable()) {
                         retry(blockException);
-                        return false;
+                        return;
                     } else {
                         throw blockException;
                     }
@@ -138,13 +140,14 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                 request.concreteIndex(indexNameExpressionResolver.concreteSingleIndex(observer.observedState(), request));
                 // check if we need to execute, and if not, return
                 if (!resolveRequest(observer.observedState(), request, listener)) {
-                    return true;
+                    listener.onFailure(new IllegalStateException(LoggerMessageFormat.format("{} request {} could not be resolved", new ShardId(request.index, request.shardId), actionName)));
+                    return;
                 }
                 blockException = checkRequestBlock(observer.observedState(), request);
                 if (blockException != null) {
                     if (blockException.retryable()) {
                         retry(blockException);
-                        return false;
+                        return;
                     } else {
                         throw blockException;
                     }
@@ -152,13 +155,13 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                 shardIt = shards(observer.observedState(), request);
             } catch (Throwable e) {
                 listener.onFailure(e);
-                return true;
+                return;
             }
 
             // no shardIt, might be in the case between index gateway recovery and shardIt initialization
             if (shardIt.size() == 0) {
                 retry(null);
-                return false;
+                return;
             }
 
             // this transport only make sense with an iterator that returns a single shard routing (like primary)
@@ -169,11 +172,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
 
             if (!shard.active()) {
                 retry(null);
-                return false;
-            }
-
-            if (!operationStarted.compareAndSet(false, true)) {
-                return true;
+                return;
             }
 
             request.shardId = shardIt.shardId().id();
@@ -197,24 +196,30 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
 
                 @Override
                 public void handleException(TransportException exp) {
+                    Throwable cause = exp.unwrapCause();
                     // 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 ||
+                    if (cause instanceof ConnectTransportException || cause 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);
+                        retry(cause);
                     } else {
                         listener.onFailure(exp);
                     }
                 }
             });
-            return true;
         }
 
         void retry(final @Nullable Throwable failure) {
             if (observer.isTimedOut()) {
                 // we running as a last attempt after a timeout has happened. don't retry
+                Throwable listenFailure = failure;
+                if (listenFailure == null) {
+                    if (shardIt == null) {
+                        listenFailure = new UnavailableShardsException(new ShardId(request.concreteIndex(), -1), "Timeout waiting for [{}], request: {}", request.timeout(), actionName);
+                    } else {
+                        listenFailure = new UnavailableShardsException(shardIt.shardId(), "[{}] shardIt, [{}] active : Timeout waiting for [{}], request: {}", shardIt.size(), shardIt.sizeActive(), request.timeout(), actionName);
+                    }
+                }
+                listener.onFailure(listenFailure);
                 return;
             }
 
@@ -232,17 +237,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                 @Override
                 public void onTimeout(TimeValue timeout) {
                     // just to be on the safe side, see if we can start it now?
-                    if (!doStart()) {
-                        Throwable listenFailure = failure;
-                        if (listenFailure == null) {
-                            if (shardIt == null) {
-                                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: " + request.toString());
-                            }
-                        }
-                        listener.onFailure(listenFailure);
-                    }
+                    doStart();
                 }
             }, request.timeout());
         }
diff --git a/core/src/test/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationActionTests.java b/core/src/test/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationActionTests.java
new file mode 100644
index 00000000000..fce431238dd
--- /dev/null
+++ b/core/src/test/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationActionTests.java
@@ -0,0 +1,316 @@
+/*
+ * 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.support.single.instance;
+
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.ActionFilter;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.replication.ClusterStateCreationUtils;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlock;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.block.ClusterBlocks;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.ShardIterator;
+import org.elasticsearch.cluster.routing.ShardRoutingState;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.cluster.TestClusterService;
+import org.elasticsearch.test.transport.CapturingTransport;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.ConnectTransportException;
+import org.elasticsearch.transport.TransportException;
+import org.elasticsearch.transport.TransportService;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+import static org.hamcrest.core.IsEqual.equalTo;
+
+public class TransportInstanceSingleOperationActionTests extends ESTestCase {
+
+    private static ThreadPool THREAD_POOL;
+
+    private TestClusterService clusterService;
+    private CapturingTransport transport;
+    private TransportService transportService;
+
+    private TestTransportInstanceSingleOperationAction action;
+
+    public static class Request extends InstanceShardOperationRequest<Request> {
+        public Request() {
+        }
+    }
+
+    public static class Response extends ActionResponse {
+        public Response() {
+        }
+    }
+
+    class TestTransportInstanceSingleOperationAction extends TransportInstanceSingleOperationAction<Request, Response> {
+        private final Map<ShardId, Object> shards = new HashMap<>();
+
+        public TestTransportInstanceSingleOperationAction(Settings settings, String actionName, TransportService transportService, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver, Supplier<Request> request) {
+            super(settings, actionName, THREAD_POOL, TransportInstanceSingleOperationActionTests.this.clusterService, transportService, actionFilters, indexNameExpressionResolver, request);
+        }
+
+        public Map<ShardId, Object> getResults() {
+            return shards;
+        }
+
+        @Override
+        protected String executor() {
+            return ThreadPool.Names.SAME;
+        }
+
+        @Override
+        protected void shardOperation(Request request, ActionListener<Response> listener) {
+            throw new UnsupportedOperationException("Not implemented in test class");
+        }
+
+        @Override
+        protected Response newResponse() {
+            return new Response();
+        }
+
+        @Override
+        protected boolean resolveRequest(ClusterState state, Request request, ActionListener<Response> listener) {
+            return true;
+        }
+
+        @Override
+        protected ShardIterator shards(ClusterState clusterState, Request request) {
+            return clusterState.routingTable().index(request.concreteIndex()).shard(request.shardId).primaryShardIt();
+        }
+    }
+
+    class MyResolver extends IndexNameExpressionResolver {
+        public MyResolver() {
+            super(Settings.EMPTY);
+        }
+
+        @Override
+        public String[] concreteIndices(ClusterState state, IndicesRequest request) {
+            return request.indices();
+        }
+    }
+
+    @BeforeClass
+    public static void startThreadPool() {
+        THREAD_POOL = new ThreadPool(TransportInstanceSingleOperationActionTests.class.getSimpleName());
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+        transport = new CapturingTransport();
+        clusterService = new TestClusterService(THREAD_POOL);
+        transportService = new TransportService(transport, THREAD_POOL);
+        transportService.start();
+        action = new TestTransportInstanceSingleOperationAction(
+                Settings.EMPTY,
+                "indices:admin/test",
+                transportService,
+                new ActionFilters(new HashSet<ActionFilter>()),
+                new MyResolver(),
+                Request::new
+        );
+    }
+
+    @AfterClass
+    public static void destroyThreadPool() {
+        ThreadPool.terminate(THREAD_POOL, 30, TimeUnit.SECONDS);
+        // since static must set to null to be eligible for collection
+        THREAD_POOL = null;
+    }
+
+    public void testGlobalBlock() {
+        Request request = new Request();
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        ClusterBlocks.Builder block = ClusterBlocks.builder()
+                .addGlobalBlock(new ClusterBlock(1, "", false, true, RestStatus.SERVICE_UNAVAILABLE, ClusterBlockLevel.ALL));
+        clusterService.setState(ClusterState.builder(clusterService.state()).blocks(block));
+        try {
+            action.new AsyncSingleAction(request, listener).start();
+            listener.get();
+            fail("expected ClusterBlockException");
+        } catch (Throwable t) {
+            if (ExceptionsHelper.unwrap(t, ClusterBlockException.class) == null) {
+                logger.info("expected ClusterBlockException  but got ", t);
+                fail("expected ClusterBlockException");
+            }
+        }
+    }
+
+    public void testBasicRequestWorks() throws InterruptedException, ExecutionException, TimeoutException {
+        Request request = new Request().index("test");
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        clusterService.setState(ClusterStateCreationUtils.state("test", randomBoolean(), ShardRoutingState.STARTED));
+        action.new AsyncSingleAction(request, listener).start();
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        transport.handleResponse(transport.capturedRequests()[0].requestId, new Response());
+        listener.get();
+    }
+
+    public void testFailureWithoutRetry() throws Exception {
+        Request request = new Request().index("test");
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        clusterService.setState(ClusterStateCreationUtils.state("test", randomBoolean(), ShardRoutingState.STARTED));
+
+        action.new AsyncSingleAction(request, listener).start();
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        long requestId = transport.capturedRequests()[0].requestId;
+        transport.clear();
+        // this should not trigger retry or anything and the listener should report exception immediately
+        transport.handleResponse(requestId, new TransportException("a generic transport exception", new Exception("generic test exception")));
+
+        try {
+            // result should return immediately
+            assertTrue(listener.isDone());
+            listener.get();
+            fail("this should fail with a transport exception");
+        } catch (ExecutionException t) {
+            if (ExceptionsHelper.unwrap(t, TransportException.class) == null) {
+                logger.info("expected TransportException  but got ", t);
+                fail("expected and TransportException");
+            }
+        }
+    }
+
+    public void testSuccessAfterRetryWithClusterStateUpdate() throws Exception {
+        Request request = new Request().index("test");
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        boolean local = randomBoolean();
+        clusterService.setState(ClusterStateCreationUtils.state("test", local, ShardRoutingState.INITIALIZING));
+        action.new AsyncSingleAction(request, listener).start();
+        // this should fail because primary not initialized
+        assertThat(transport.capturedRequests().length, equalTo(0));
+        clusterService.setState(ClusterStateCreationUtils.state("test", local, ShardRoutingState.STARTED));
+        // this time it should work
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        transport.handleResponse(transport.capturedRequests()[0].requestId, new Response());
+        listener.get();
+    }
+
+    public void testSuccessAfterRetryWithExcpetionFromTransport() throws Exception {
+        Request request = new Request().index("test");
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        boolean local = randomBoolean();
+        clusterService.setState(ClusterStateCreationUtils.state("test", local, ShardRoutingState.STARTED));
+        action.new AsyncSingleAction(request, listener).start();
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        long requestId = transport.capturedRequests()[0].requestId;
+        transport.clear();
+        DiscoveryNode node = clusterService.state().getNodes().getLocalNode();
+        transport.handleResponse(requestId, new ConnectTransportException(node, "test exception"));
+        // trigger cluster state observer
+        clusterService.setState(ClusterStateCreationUtils.state("test", local, ShardRoutingState.STARTED));
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        transport.handleResponse(transport.capturedRequests()[0].requestId, new Response());
+        listener.get();
+    }
+
+    public void testRetryOfAnAlreadyTimedOutRequest() throws Exception {
+        Request request = new Request().index("test").timeout(new TimeValue(0, TimeUnit.MILLISECONDS));
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        clusterService.setState(ClusterStateCreationUtils.state("test", randomBoolean(), ShardRoutingState.STARTED));
+        action.new AsyncSingleAction(request, listener).start();
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        long requestId = transport.capturedRequests()[0].requestId;
+        transport.clear();
+        DiscoveryNode node = clusterService.state().getNodes().getLocalNode();
+        transport.handleResponse(requestId, new ConnectTransportException(node, "test exception"));
+
+        // wait until the timeout was triggered and we actually tried to send for the second time
+        assertBusy(new Runnable() {
+            @Override
+            public void run() {
+                assertThat(transport.capturedRequests().length, equalTo(1));
+            }
+        });
+
+        // let it fail the second time too
+        requestId = transport.capturedRequests()[0].requestId;
+        transport.handleResponse(requestId, new ConnectTransportException(node, "test exception"));
+        try {
+            // result should return immediately
+            assertTrue(listener.isDone());
+            listener.get();
+            fail("this should fail with a transport exception");
+        } catch (ExecutionException t) {
+            if (ExceptionsHelper.unwrap(t, ConnectTransportException.class) == null) {
+                logger.info("expected ConnectTransportException  but got ", t);
+                fail("expected and ConnectTransportException");
+            }
+        }
+    }
+
+    public void testUnresolvableRequestDoesNotHang() throws InterruptedException, ExecutionException, TimeoutException {
+        action = new TestTransportInstanceSingleOperationAction(
+                Settings.EMPTY,
+                "indices:admin/test_unresolvable",
+                transportService,
+                new ActionFilters(new HashSet<ActionFilter>()),
+                new MyResolver(),
+                Request::new
+        ) {
+            @Override
+            protected boolean resolveRequest(ClusterState state, Request request, ActionListener<Response> listener) {
+                return false;
+            }
+        };
+        Request request = new Request().index("test");
+        request.shardId = 0;
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+        clusterService.setState(ClusterStateCreationUtils.state("test", randomBoolean(), ShardRoutingState.STARTED));
+        action.new AsyncSingleAction(request, listener).start();
+        assertThat(transport.capturedRequests().length, equalTo(0));
+        try {
+            listener.get();
+        } catch (Throwable t) {
+            if (ExceptionsHelper.unwrap(t, IllegalStateException.class) == null) {
+                logger.info("expected IllegalStateException  but got ", t);
+                fail("expected and IllegalStateException");
+            }
+        }
+    }
+}
\ No newline at end of file