diff --git a/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java b/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java
index b0d98c39454..72dc8db03a6 100644
--- a/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java
+++ b/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java
@@ -336,14 +336,18 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
 
                     if (updateTask instanceof AckedClusterStateUpdateTask) {
                         final AckedClusterStateUpdateTask ackedUpdateTask = (AckedClusterStateUpdateTask) updateTask;
-                        try {
-                            ackListener = new AckCountDownListener(ackedUpdateTask, newClusterState.version(), newClusterState.nodes(), threadPool);
-                        } catch(EsRejectedExecutionException ex) {
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("Couldn't schedule timeout thread - node might be shutting down", ex);
-                            }
-                            //timeout straightaway, otherwise we could wait forever as the timeout thread has not started
+                        if (ackedUpdateTask.ackTimeout() == null || ackedUpdateTask.ackTimeout().millis() == 0) {
                             ackedUpdateTask.onAckTimeout();
+                        } else {
+                            try {
+                                ackListener = new AckCountDownListener(ackedUpdateTask, newClusterState, threadPool);
+                            } catch(EsRejectedExecutionException ex) {
+                                if (logger.isDebugEnabled()) {
+                                    logger.debug("Couldn't schedule timeout thread - node might be shutting down", ex);
+                                }
+                                //timeout straightaway, otherwise we could wait forever as the timeout thread has not started
+                                ackedUpdateTask.onAckTimeout();
+                            }
                         }
                     }
                 } else {
@@ -624,21 +628,23 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
 
     private class AckCountDownListener implements Discovery.AckListener {
         private final AckedClusterStateUpdateTask ackedUpdateTask;
-        private final long version;
         private final CountDown countDown;
+        private final ClusterState clusterState;
         private final Future<?> ackTimeoutCallback;
         private Throwable lastFailure;
 
-        AckCountDownListener(AckedClusterStateUpdateTask ackedUpdateTask, long clusterStateVersion, DiscoveryNodes nodes, ThreadPool threadPool) {
+        AckCountDownListener(AckedClusterStateUpdateTask ackedUpdateTask, ClusterState clusterState, ThreadPool threadPool) {
             this.ackedUpdateTask = ackedUpdateTask;
-            this.version = clusterStateVersion;
+            this.clusterState = clusterState;
             int countDown = 0;
-            for (DiscoveryNode node : nodes) {
+            for (DiscoveryNode node : clusterState.nodes()) {
                 if (ackedUpdateTask.mustAck(node)) {
                     countDown++;
                 }
             }
-            logger.trace("expecting {} acknowledgements for cluster_state update (version: {})", countDown, version);
+            //we always wait for at least 1 node (the master)
+            countDown = Math.max(1, countDown);
+            logger.trace("expecting {} acknowledgements for cluster_state update (version: {})", countDown, clusterState.version());
             this.countDown = new CountDown(countDown);
             this.ackTimeoutCallback = threadPool.schedule(ackedUpdateTask.ackTimeout(), ThreadPool.Names.GENERIC, new Runnable() {
                 @Override
@@ -651,17 +657,20 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
         @Override
         public void onNodeAck(DiscoveryNode node, @Nullable Throwable t) {
             if (!ackedUpdateTask.mustAck(node)) {
-                return;
+                //we always wait for the master ack anyway
+                if (!node.equals(clusterState.nodes().masterNode())) {
+                    return;
+                }
             }
             if (t == null) {
-                logger.trace("ack received from node [{}], cluster_state update (version: {})", node, version);
+                logger.trace("ack received from node [{}], cluster_state update (version: {})", node, clusterState.version());
             } else {
                 this.lastFailure = t;
-                logger.debug("ack received from node [{}], cluster_state update (version: {})", t, node, version);
+                logger.debug("ack received from node [{}], cluster_state update (version: {})", t, node, clusterState.version());
             }
 
             if (countDown.countDown()) {
-                logger.trace("all expected nodes acknowledged cluster_state update (version: {})", version);
+                logger.trace("all expected nodes acknowledged cluster_state update (version: {})", clusterState.version());
                 ackTimeoutCallback.cancel(true);
                 ackedUpdateTask.onAllNodesAcked(lastFailure);
             }
@@ -670,7 +679,7 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
         @Override
         public void onTimeout() {
             if (countDown.fastForward()) {
-                logger.trace("timeout waiting for acknowledgement for cluster_state update (version: {})", version);
+                logger.trace("timeout waiting for acknowledgement for cluster_state update (version: {})", clusterState.version());
                 ackedUpdateTask.onAckTimeout();
             }
         }
diff --git a/src/test/java/org/elasticsearch/cluster/ClusterServiceTests.java b/src/test/java/org/elasticsearch/cluster/ClusterServiceTests.java
index 3ae16a8a0df..ebf9b6f63ec 100644
--- a/src/test/java/org/elasticsearch/cluster/ClusterServiceTests.java
+++ b/src/test/java/org/elasticsearch/cluster/ClusterServiceTests.java
@@ -22,7 +22,9 @@ package org.elasticsearch.cluster;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksResponse;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.PendingClusterTask;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.component.LifecycleComponent;
@@ -109,6 +111,287 @@ public class ClusterServiceTests extends AbstractIntegrationTest {
         assertThat(executeCalled.get(), equalTo(false));
     }
 
+    @Test
+    public void testAckedUpdateTask() throws Exception {
+        Settings settings = settingsBuilder()
+                .put("discovery.type", "zen")
+                .put("discovery.zen.minimum_master_nodes", 1)
+                .put("discovery.zen.ping_timeout", "200ms")
+                .put("discovery.initial_state_timeout", "500ms")
+                .build();
+        cluster().startNode(settings);
+        ClusterService clusterService = cluster().getInstance(ClusterService.class);
+
+        final AtomicBoolean allNodesAcked = new AtomicBoolean(false);
+        final AtomicBoolean ackTimeout = new AtomicBoolean(false);
+        final AtomicBoolean onFailure = new AtomicBoolean(false);
+        final AtomicBoolean executed = new AtomicBoolean(false);
+        final AtomicBoolean processed = new AtomicBoolean(false);
+        final CountDownLatch latch = new CountDownLatch(1);
+        clusterService.submitStateUpdateTask("test", new AckedClusterStateUpdateTask() {
+            @Override
+            public boolean mustAck(DiscoveryNode discoveryNode) {
+                return true;
+            }
+
+            @Override
+            public void onAllNodesAcked(@Nullable Throwable t) {
+                allNodesAcked.set(true);
+            }
+
+            @Override
+            public void onAckTimeout() {
+                ackTimeout.set(true);
+            }
+
+            @Override
+            public TimeValue ackTimeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public TimeValue timeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                processed.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                executed.set(true);
+                return ClusterState.newClusterStateBuilder().state(currentState).build();
+            }
+
+            @Override
+            public void onFailure(String source, Throwable t) {
+                onFailure.set(true);
+                latch.countDown();
+            }
+        });
+
+        assertThat(latch.await(1, TimeUnit.SECONDS), equalTo(true));
+
+        assertThat(allNodesAcked.get(), equalTo(true));
+        assertThat(ackTimeout.get(), equalTo(false));
+        assertThat(executed.get(), equalTo(true));
+        assertThat(processed.get(), equalTo(true));
+        assertThat(onFailure.get(), equalTo(false));
+    }
+
+    @Test
+    public void testAckedUpdateTaskSameClusterState() throws Exception {
+        Settings settings = settingsBuilder()
+                .put("discovery.type", "zen")
+                .put("discovery.zen.minimum_master_nodes", 1)
+                .put("discovery.zen.ping_timeout", "200ms")
+                .put("discovery.initial_state_timeout", "500ms")
+                .build();
+        cluster().startNode(settings);
+        ClusterService clusterService = cluster().getInstance(ClusterService.class);
+
+        final AtomicBoolean allNodesAcked = new AtomicBoolean(false);
+        final AtomicBoolean ackTimeout = new AtomicBoolean(false);
+        final AtomicBoolean onFailure = new AtomicBoolean(false);
+        final AtomicBoolean executed = new AtomicBoolean(false);
+        final AtomicBoolean processed = new AtomicBoolean(false);
+        final CountDownLatch latch = new CountDownLatch(1);
+        clusterService.submitStateUpdateTask("test", new AckedClusterStateUpdateTask() {
+            @Override
+            public boolean mustAck(DiscoveryNode discoveryNode) {
+                return true;
+            }
+
+            @Override
+            public void onAllNodesAcked(@Nullable Throwable t) {
+                allNodesAcked.set(true);
+            }
+
+            @Override
+            public void onAckTimeout() {
+                ackTimeout.set(true);
+            }
+
+            @Override
+            public TimeValue ackTimeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public TimeValue timeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                processed.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                executed.set(true);
+                return currentState;
+            }
+
+            @Override
+            public void onFailure(String source, Throwable t) {
+                onFailure.set(true);
+                latch.countDown();
+            }
+        });
+
+        assertThat(latch.await(1, TimeUnit.SECONDS), equalTo(true));
+
+        assertThat(allNodesAcked.get(), equalTo(true));
+        assertThat(ackTimeout.get(), equalTo(false));
+        assertThat(executed.get(), equalTo(true));
+        assertThat(processed.get(), equalTo(true));
+        assertThat(onFailure.get(), equalTo(false));
+    }
+
+    @Test
+    public void testAckedUpdateTaskNoAckExpected() throws Exception {
+        Settings settings = settingsBuilder()
+                .put("discovery.type", "zen")
+                .put("discovery.zen.minimum_master_nodes", 1)
+                .put("discovery.zen.ping_timeout", "200ms")
+                .put("discovery.initial_state_timeout", "500ms")
+                .build();
+        cluster().startNode(settings);
+        ClusterService clusterService = cluster().getInstance(ClusterService.class);
+
+        final AtomicBoolean allNodesAcked = new AtomicBoolean(false);
+        final AtomicBoolean ackTimeout = new AtomicBoolean(false);
+        final AtomicBoolean onFailure = new AtomicBoolean(false);
+        final AtomicBoolean executed = new AtomicBoolean(false);
+        final CountDownLatch latch = new CountDownLatch(1);
+        clusterService.submitStateUpdateTask("test", new AckedClusterStateUpdateTask() {
+            @Override
+            public boolean mustAck(DiscoveryNode discoveryNode) {
+                return false;
+            }
+
+            @Override
+            public void onAllNodesAcked(@Nullable Throwable t) {
+                allNodesAcked.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public void onAckTimeout() {
+                ackTimeout.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public TimeValue ackTimeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public TimeValue timeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+            }
+
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                executed.set(true);
+                return ClusterState.newClusterStateBuilder().state(currentState).build();
+            }
+
+            @Override
+            public void onFailure(String source, Throwable t) {
+                onFailure.set(true);
+                latch.countDown();
+            }
+        });
+
+        assertThat(latch.await(1, TimeUnit.SECONDS), equalTo(true));
+
+        assertThat(allNodesAcked.get(), equalTo(true));
+        assertThat(ackTimeout.get(), equalTo(false));
+        assertThat(executed.get(), equalTo(true));
+        assertThat(onFailure.get(), equalTo(false));
+    }
+
+    @Test
+    public void testAckedUpdateTaskTimeoutZero() throws Exception {
+        Settings settings = settingsBuilder()
+                .put("discovery.type", "zen")
+                .put("discovery.zen.minimum_master_nodes", 1)
+                .put("discovery.zen.ping_timeout", "200ms")
+                .put("discovery.initial_state_timeout", "500ms")
+                .build();
+        cluster().startNode(settings);
+        ClusterService clusterService = cluster().getInstance(ClusterService.class);
+
+        final AtomicBoolean allNodesAcked = new AtomicBoolean(false);
+        final AtomicBoolean ackTimeout = new AtomicBoolean(false);
+        final AtomicBoolean onFailure = new AtomicBoolean(false);
+        final AtomicBoolean executed = new AtomicBoolean(false);
+        final CountDownLatch latch = new CountDownLatch(1);
+        clusterService.submitStateUpdateTask("test", new AckedClusterStateUpdateTask() {
+            @Override
+            public boolean mustAck(DiscoveryNode discoveryNode) {
+                return false;
+            }
+
+            @Override
+            public void onAllNodesAcked(@Nullable Throwable t) {
+                allNodesAcked.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public void onAckTimeout() {
+                ackTimeout.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public TimeValue ackTimeout() {
+                return TimeValue.timeValueSeconds(0);
+            }
+
+            @Override
+            public TimeValue timeout() {
+                return TimeValue.timeValueSeconds(10);
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+
+            }
+
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                executed.set(true);
+                return ClusterState.newClusterStateBuilder().state(currentState).build();
+            }
+
+            @Override
+            public void onFailure(String source, Throwable t) {
+                onFailure.set(true);
+                latch.countDown();
+            }
+        });
+
+        assertThat(latch.await(1, TimeUnit.SECONDS), equalTo(true));
+
+        assertThat(allNodesAcked.get(), equalTo(false));
+        assertThat(ackTimeout.get(), equalTo(true));
+        assertThat(executed.get(), equalTo(true));
+        assertThat(onFailure.get(), equalTo(false));
+    }
+
     @Test
     public void testPendingUpdateTask() throws Exception {
         Settings zenSettings = settingsBuilder()
diff --git a/src/test/java/org/elasticsearch/cluster/ack/AckTests.java b/src/test/java/org/elasticsearch/cluster/ack/AckTests.java
index 20a6b62e411..3a2d385d1b5 100644
--- a/src/test/java/org/elasticsearch/cluster/ack/AckTests.java
+++ b/src/test/java/org/elasticsearch/cluster/ack/AckTests.java
@@ -74,6 +74,15 @@ public class AckTests extends AbstractIntegrationTest {
         }
     }
 
+    @Test
+    public void testUpdateSettingsNoAcknowledgement() {
+        createIndex("test");
+
+        UpdateSettingsResponse updateSettingsResponse = client().admin().indices().prepareUpdateSettings("test").setTimeout("0s")
+                .setSettings(ImmutableSettings.builder().put("refresh_interval", 9999)).get();
+        assertThat(updateSettingsResponse.isAcknowledged(), equalTo(false));
+    }
+
     @Test
     public void testPutWarmerAcknowledgement() {
         createIndex("test");
@@ -94,6 +103,17 @@ public class AckTests extends AbstractIntegrationTest {
         }
     }
 
+    @Test
+    public void testPutWarmerNoAcknowledgement() {
+        createIndex("test");
+        ensureGreen();
+
+        PutWarmerResponse putWarmerResponse = client().admin().indices().preparePutWarmer("custom_warmer").setTimeout("0s")
+                .setSearchRequest(client().prepareSearch("test").setTypes("test").setQuery(QueryBuilders.matchAllQuery()))
+                .get();
+        assertThat(putWarmerResponse.isAcknowledged(), equalTo(false));
+    }
+
     @Test
     public void testDeleteWarmerAcknowledgement() {
         createIndex("test");
@@ -113,6 +133,17 @@ public class AckTests extends AbstractIntegrationTest {
         }
     }
 
+    @Test
+    public void testDeleteWarmerNoAcknowledgement() {
+        createIndex("test");
+        ensureGreen();
+
+        PutWarmerResponse putWarmerResponse = client().admin().indices().preparePutWarmer("custom_warmer").setTimeout("0s")
+                .setSearchRequest(client().prepareSearch("test").setTypes("test").setQuery(QueryBuilders.matchAllQuery()))
+                .get();
+        assertThat(putWarmerResponse.isAcknowledged(), equalTo(false));
+    }
+
     @Test
     public void testDeleteMappingAcknowledgement() {
         client().admin().indices().prepareCreate("test")
@@ -133,6 +164,18 @@ public class AckTests extends AbstractIntegrationTest {
         }
     }
 
+    @Test
+    public void testDeleteMappingNoAcknowledgement() {
+        client().admin().indices().prepareCreate("test")
+                .addMapping("type1", "field1", "type=string").get();
+        ensureGreen();
+
+        client().prepareIndex("test", "type1").setSource("field1", "value1");
+
+        DeleteMappingResponse deleteMappingResponse = client().admin().indices().prepareDeleteMapping("test").setTimeout("0s").setType("type1").get();
+        assertThat(deleteMappingResponse.isAcknowledged(), equalTo(false));
+    }
+
     @Test
     public void testClusterRerouteAcknowledgement() throws InterruptedException {
         client().admin().indices().prepareCreate("test")
@@ -173,6 +216,20 @@ public class AckTests extends AbstractIntegrationTest {
         waitForRelocation();
     }
 
+    @Test
+    public void testClusterRerouteNoAcknowledgement() throws InterruptedException {
+        client().admin().indices().prepareCreate("test")
+                .setSettings(settingsBuilder()
+                        .put("number_of_shards", atLeast(cluster().numNodes()))
+                        .put("number_of_replicas", 0)).get();
+        ensureGreen();
+
+        MoveAllocationCommand moveAllocationCommand = getAllocationCommand();
+
+        ClusterRerouteResponse clusterRerouteResponse = client().admin().cluster().prepareReroute().setTimeout("0s").add(moveAllocationCommand).get();
+        assertThat(clusterRerouteResponse.isAcknowledged(), equalTo(false));
+    }
+
     @Test
     public void testClusterRerouteAcknowledgementDryRun() throws InterruptedException {
         client().admin().indices().prepareCreate("test")
@@ -209,6 +266,21 @@ public class AckTests extends AbstractIntegrationTest {
         }
     }
 
+    @Test
+    public void testClusterRerouteNoAcknowledgementDryRun() throws InterruptedException {
+        client().admin().indices().prepareCreate("test")
+                .setSettings(settingsBuilder()
+                        .put("number_of_shards", atLeast(cluster().numNodes()))
+                        .put("number_of_replicas", 0)).get();
+        ensureGreen();
+
+        MoveAllocationCommand moveAllocationCommand = getAllocationCommand();
+
+        ClusterRerouteResponse clusterRerouteResponse = client().admin().cluster().prepareReroute().setTimeout("0s").setDryRun(true).add(moveAllocationCommand).get();
+        //acknowledged anyway as no changes were made
+        assertThat(clusterRerouteResponse.isAcknowledged(), equalTo(true));
+    }
+
     private MoveAllocationCommand getAllocationCommand() {
         String fromNodeId = null;
         String toNodeId = null;
@@ -281,4 +353,34 @@ public class AckTests extends AbstractIntegrationTest {
         //removes the allocation exclude settings
         client().admin().cluster().prepareUpdateSettings().setTransientSettings(settingsBuilder().put("cluster.routing.allocation.exclude._id", "")).get();
     }
+
+    @Test
+    public void testClusterUpdateSettingsNoAcknowledgement() {
+        client().admin().indices().prepareCreate("test")
+                .setSettings(settingsBuilder()
+                        .put("number_of_shards", atLeast(cluster().numNodes()))
+                        .put("number_of_replicas", 0)).get();
+        ensureGreen();
+
+        NodesInfoResponse nodesInfo = client().admin().cluster().prepareNodesInfo().get();
+        String excludedNodeId = null;
+        for (NodeInfo nodeInfo : nodesInfo) {
+            if (nodeInfo.getNode().isDataNode()) {
+                excludedNodeId = nodesInfo.getAt(0).getNode().id();
+                break;
+            }
+        }
+        assert excludedNodeId != null;
+
+        ClusterUpdateSettingsResponse clusterUpdateSettingsResponse = client().admin().cluster().prepareUpdateSettings().setTimeout("0s")
+                .setTransientSettings(settingsBuilder().put("cluster.routing.allocation.exclude._id", excludedNodeId)).get();
+        assertThat(clusterUpdateSettingsResponse.isAcknowledged(), equalTo(false));
+        assertThat(clusterUpdateSettingsResponse.getTransientSettings().get("cluster.routing.allocation.exclude._id"), equalTo(excludedNodeId));
+
+        //let's wait for the relocation to be completed, otherwise there can be issues with after test checks (mock directory wrapper etc.)
+        waitForRelocation();
+
+        //removes the allocation exclude settings
+        client().admin().cluster().prepareUpdateSettings().setTransientSettings(settingsBuilder().put("cluster.routing.allocation.exclude._id", "")).get();
+    }
 }