Cluster Health should run on applied states, even if waitFor=0

core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -143,7 +143,7 @@ private void executeHealth(final ClusterHealthRequest request, final ActionListe
         assert waitFor >= 0;
         final ClusterStateObserver observer = new ClusterStateObserver(clusterService, logger, threadPool.getThreadContext());
         final ClusterState state = observer.observedState();
-        if (waitFor == 0 || request.timeout().millis() == 0) {
+        if (request.timeout().millis() == 0) {
             listener.onResponse(getResponse(request, state, waitFor, request.timeout().millis() == 0));
             return;
         }

core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
-import org.elasticsearch.action.support.ThreadedActionListener;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.MasterNodeChangePredicate;
@@ -116,10 +115,6 @@ protected boolean validate(ClusterState newState) {
             if (task != null) {
                 request.setParentTask(clusterService.localNode().getId(), task.getId());
             }
-            // TODO do we really need to wrap it in a listener? the handlers should be cheap
-            if ((listener instanceof ThreadedActionListener) == false) {
-                listener = new ThreadedActionListener<>(logger, threadPool, ThreadPool.Names.LISTENER, listener);
-            }
             this.listener = listener;
         }
 

core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTookTests.java

@@ -41,7 +41,9 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
+import org.junit.BeforeClass;
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashSet;
@@ -53,17 +55,26 @@
 import static org.elasticsearch.test.StreamsUtils.copyToStringFromClasspath;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.mockito.Mockito.mock;
 
 public class TransportBulkActionTookTests extends ESTestCase {
 
-    private ThreadPool threadPool;
+    static private ThreadPool threadPool;
     private ClusterService clusterService;
 
+    @BeforeClass
+    public static void beforeClass() {
+        threadPool = new ThreadPool("TransportBulkActionTookTests");
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        ThreadPool.terminate(threadPool, 30, TimeUnit.SECONDS);
+        threadPool = null;
+    }
+
     @Before
     public void setUp() throws Exception {
         super.setUp();
-        threadPool = mock(ThreadPool.class);
         clusterService = createClusterService(threadPool);
     }
 

core/src/test/java/org/elasticsearch/cluster/health/ClusterStateHealthTests.java

@@ -19,22 +19,42 @@
 package org.elasticsearch.cluster.health;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.cluster.health.TransportClusterHealthAction;
+import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.gateway.NoopGatewayAllocator;
+import org.elasticsearch.test.transport.CapturingTransport;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
 import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
 
 import java.io.IOException;
+import java.util.HashSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.cluster.service.ClusterServiceUtils.createClusterService;
 import static org.hamcrest.CoreMatchers.allOf;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.empty;
@@ -45,6 +65,83 @@
 public class ClusterStateHealthTests extends ESTestCase {
     private final IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver(Settings.EMPTY);
 
+
+    private static ThreadPool threadPool;
+
+    private ClusterService clusterService;
+    private TransportService transportService;
+    private CapturingTransport transport;
+
+    @BeforeClass
+    public static void beforeClass() {
+        threadPool = new ThreadPool("ClusterStateHealthTests");
+    }
+
+    @Override
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+        transport = new CapturingTransport();
+        clusterService = createClusterService(threadPool);
+        transportService = new TransportService(transport, threadPool);
+        transportService.start();
+        transportService.acceptIncomingRequests();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        super.tearDown();
+        clusterService.close();
+        transportService.close();
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        ThreadPool.terminate(threadPool, 30, TimeUnit.SECONDS);
+        threadPool = null;
+    }
+
+    public void testClusterHealthWaitsForClusterStateApplication() throws InterruptedException, ExecutionException {
+        final CountDownLatch applyLatch = new CountDownLatch(1);
+        final CountDownLatch listenerCalled = new CountDownLatch(1);
+        clusterService.add(event -> {
+            listenerCalled.countDown();
+            try {
+                applyLatch.await();
+            } catch (InterruptedException e) {
+                logger.debug("interrupted", e);
+            }
+        });
+
+        clusterService.submitStateUpdateTask("test", new ClusterStateUpdateTask() {
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                return ClusterState.builder(currentState).build();
+            }
+
+            @Override
+            public void onFailure(String source, Throwable t) {
+                logger.warn("unexpected failure", t);
+            }
+        });
+
+        logger.info("--> waiting for listener to be called and cluster state being blocked");
+        listenerCalled.await();
+
+        TransportClusterHealthAction action = new TransportClusterHealthAction(Settings.EMPTY, transportService,
+            clusterService, threadPool, clusterService.state().getClusterName(), new ActionFilters(new HashSet<>()),
+            indexNameExpressionResolver, NoopGatewayAllocator.INSTANCE);
+        PlainActionFuture<ClusterHealthResponse> listener = new PlainActionFuture<>();
+
+        action.execute(new ClusterHealthRequest(), listener);
+
+        assertFalse(listener.isDone());
+
+        applyLatch.countDown();
+        listener.get();
+    }
+
+
     public void testClusterHealth() throws IOException {
         RoutingTableGenerator routingTableGenerator = new RoutingTableGenerator();
         RoutingTableGenerator.ShardCounter counter = new RoutingTableGenerator.ShardCounter();

core/src/test/java/org/elasticsearch/cluster/routing/allocation/ShardStateIT.java

@@ -48,7 +48,8 @@ public void testPrimaryFailureIncreasesTerm() throws Exception {
         indicesService.indexService(resolveIndex("test")).getShard(shard).failShard("simulated test failure", null);
 
         logger.info("--> waiting for a yellow index");
-        assertBusy(() -> assertThat(client().admin().cluster().prepareHealth().get().getStatus(), equalTo(ClusterHealthStatus.YELLOW)));
+        assertBusy(() -> assertThat(client().admin().cluster().prepareHealth().get().getStatus(),
+            equalTo(ClusterHealthStatus.YELLOW)));
 
         final long term0 = shard == 0 ? 2 : 1;
         final long term1 = shard == 1 ? 2 : 1;

core/src/test/java/org/elasticsearch/cluster/service/ClusterServiceUtils.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.NodeConnectionsService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.DummyTransportAddress;
@@ -60,6 +61,9 @@ public void disconnectFromRemovedNodes(ClusterChangedEvent event) {
         clusterService.setClusterStatePublisher((event, ackListener) -> {
         });
         clusterService.start();
+        final DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(clusterService.state().nodes());
+        nodes.masterNodeId(clusterService.localNode().getId());
+        setState(clusterService, ClusterState.builder(clusterService.state()).nodes(nodes));
         return clusterService;
     }