From afa0ab62261d9a46c77a8bea47679ffe5e743ddc Mon Sep 17 00:00:00 2001
From: Luca Cavanna <cavannaluca@gmail.com>
Date: Thu, 31 Oct 2013 15:42:15 +0100
Subject: [PATCH] Fixed ack behaviour when no ack is expected from any node or
 timeout is set to 0

We now return acknowledged true when no wait is needed (mustAck always returns false). We do wait for the master node to complete its actions though. Previously it would try to timeout and hang due to a CountDown#fastForward call when the internal counter is set to 0

We now return acknowledged false without starting the timeout thread when the timeout is set 0, as starting the wait and immediately stopping the thread seems pointless.

Added coverage for ack in ClusterServiceTests
---
 .../service/InternalClusterService.java       |  43 +--
 .../cluster/ClusterServiceTests.java          | 283 ++++++++++++++++++
 .../elasticsearch/cluster/ack/AckTests.java   | 102 +++++++
 3 files changed, 411 insertions(+), 17 deletions(-)

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();
+    }
 }