Use test cluster exclusive in CancellableTasksIT (#67058)

testRemoveBanParentsOnDisconnect disconnects some nodes and leaves the
cluster in an unhealthy state, which fails other tests. We can reconnect
nodes after that test. However, this commit chooses to run this suite
test-cluster exclusive instead because it is simpler and more stable.

Closes #6652

Author: dnhatn

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java

@@ -58,7 +58,7 @@
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
-import org.junit.Before;
+import org.junit.After;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -81,6 +81,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
 public class CancellableTasksIT extends ESIntegTestCase {
 
     static int idGenerator = 0;
@@ -89,13 +90,14 @@ public class CancellableTasksIT extends ESIntegTestCase {
     static final Map<TestRequest, CountDownLatch> beforeExecuteLatches = ConcurrentCollections.newConcurrentMap();
     static final Map<TestRequest, CountDownLatch> completedLatches = ConcurrentCollections.newConcurrentMap();
 
-    @Before
-    public void resetTestStates() {
-        idGenerator = 0;
-        beforeSendLatches.clear();
-        arrivedLatches.clear();
-        beforeExecuteLatches.clear();
-        completedLatches.clear();
+    @After
+    public void ensureAllBansRemoved() throws Exception {
+        assertBusy(() -> {
+            for (String node : internalCluster().getNodeNames()) {
+                TaskManager taskManager = internalCluster().getInstance(TransportService.class, node).getTaskManager();
+                assertThat("node " + node, taskManager.getBannedTaskIds(), empty());
+            }
+        }, 30, TimeUnit.SECONDS);
     }
 
     static TestRequest generateTestRequest(Set<DiscoveryNode> nodes, int level, int maxLevel) {
@@ -138,7 +140,7 @@ static Set<TestRequest> allowPartialRequest(TestRequest request) throws Exceptio
             beforeSendLatches.get(req).countDown();
         }
         for (TestRequest req : sentRequests) {
-            arrivedLatches.get(req).await();
+            assertTrue(arrivedLatches.get(req).await(60, TimeUnit.SECONDS));
         }
         Set<TestRequest> completedRequests = new HashSet<>();
         for (TestRequest req : randomSubsetOf(sentRequests)) {
@@ -151,7 +153,7 @@ static Set<TestRequest> allowPartialRequest(TestRequest request) throws Exceptio
             beforeExecuteLatches.get(req).countDown();
         }
         for (TestRequest req : completedRequests) {
-            completedLatches.get(req).await();
+            assertTrue(completedLatches.get(req).await(60, TimeUnit.SECONDS));
         }
         return Sets.difference(sentRequests, completedRequests);
     }
@@ -164,21 +166,12 @@ static void allowEntireRequest(TestRequest request) {
         }
     }
 
-    void ensureAllBansRemoved() throws Exception {
-        assertBusy(() -> {
-            for (String node : internalCluster().getNodeNames()) {
-                TaskManager taskManager = internalCluster().getInstance(TransportService.class, node).getTaskManager();
-                assertThat("node " + node, taskManager.getBannedTaskIds(), empty());
-            }
-        }, 30, TimeUnit.SECONDS);
-    }
-
     public void testBanOnlyNodesWithOutstandingDescendantTasks() throws Exception {
         if (randomBoolean()) {
-            internalCluster().startNodes(randomIntBetween(1, 3));
+            internalCluster().startNodes(1);
         }
         Set<DiscoveryNode> nodes = StreamSupport.stream(clusterService().state().nodes().spliterator(), false).collect(Collectors.toSet());
-        final TestRequest rootRequest = generateTestRequest(nodes, 0, between(1, 4));
+        final TestRequest rootRequest = generateTestRequest(nodes, 0, between(1, 3));
         ActionFuture<TestResponse> rootTaskFuture = client().execute(TransportTestAction.ACTION, rootRequest);
         Set<TestRequest> pendingRequests = allowPartialRequest(rootRequest);
         TaskId rootTaskId = getRootTaskId(rootRequest);
@@ -191,28 +184,31 @@ public void testBanOnlyNodesWithOutstandingDescendantTasks() throws Exception {
                 client().admin().cluster().prepareCancelTasks().setTaskId(subTask.getTaskId()).waitForCompletion(false).get();
             }
         }
-        assertBusy(() -> {
-            for (DiscoveryNode node : nodes) {
-                TaskManager taskManager = internalCluster().getInstance(TransportService.class, node.getName()).getTaskManager();
-                Set<TaskId> expectedBans = new HashSet<>();
-                for (TestRequest req : pendingRequests) {
-                    if (req.node.equals(node)) {
-                        List<Task> childTasks = taskManager.getTasks().values().stream()
-                            .filter(t -> t.getParentTaskId() != null && t.getDescription().equals(req.taskDescription()))
-                            .collect(Collectors.toList());
-                        assertThat(childTasks, hasSize(1));
-                        CancellableTask childTask = (CancellableTask) childTasks.get(0);
-                        assertTrue(childTask.isCancelled());
-                        expectedBans.add(childTask.getParentTaskId());
+        try {
+            assertBusy(() -> {
+                for (DiscoveryNode node : nodes) {
+                    TaskManager taskManager = internalCluster().getInstance(TransportService.class, node.getName()).getTaskManager();
+                    Set<TaskId> expectedBans = new HashSet<>();
+                    for (TestRequest req : pendingRequests) {
+                        if (req.node.equals(node)) {
+                            List<Task> childTasks = taskManager.getTasks().values().stream()
+                                .filter(t -> t.getParentTaskId() != null && t.getDescription().equals(req.taskDescription()))
+                                .collect(Collectors.toList());
+                            assertThat(childTasks, hasSize(1));
+                            CancellableTask childTask = (CancellableTask) childTasks.get(0);
+                            assertTrue(childTask.isCancelled());
+                            expectedBans.add(childTask.getParentTaskId());
+                        }
                     }
+                    assertThat(taskManager.getBannedTaskIds(), equalTo(expectedBans));
                 }
-                assertThat(taskManager.getBannedTaskIds(), equalTo(expectedBans));
-            }
-        }, 30, TimeUnit.SECONDS);
-        allowEntireRequest(rootRequest);
-        cancelFuture.actionGet();
-        waitForRootTask(rootTaskFuture);
-        ensureAllBansRemoved();
+            }, 30, TimeUnit.SECONDS);
+        } finally {
+            allowEntireRequest(rootRequest);
+            cancelFuture.actionGet();
+            waitForRootTask(rootTaskFuture);
+            ensureAllBansRemoved();
+        }
     }
 
     public void testCancelTaskMultipleTimes() throws Exception {
@@ -326,6 +322,7 @@ static void waitForRootTask(ActionFuture<TestResponse> rootTask) {
             rootTask.actionGet();
         } catch (Exception e) {
             final Throwable cause = ExceptionsHelper.unwrap(e, TaskCancelledException.class);
+            assertNotNull(cause);
             assertThat(cause.getMessage(), anyOf(
                 equalTo("The parent task was cancelled, shouldn't start any child tasks"),
                 containsString("Task cancelled before it started:"),
@@ -441,7 +438,7 @@ protected void doExecute(Task task, TestRequest request, ActionListener<TestResp
             GroupedActionListener<TestResponse> groupedListener =
                 new GroupedActionListener<>(listener.map(r -> new TestResponse()), subRequests.size() + 1);
             transportService.getThreadPool().generic().execute(ActionRunnable.supply(groupedListener, () -> {
-                beforeExecuteLatches.get(request).await();
+                assertTrue(beforeExecuteLatches.get(request).await(60, TimeUnit.SECONDS));
                 if (((CancellableTask) task).isCancelled()) {
                     throw new TaskCancelledException("Task was cancelled while executing");
                 }
@@ -465,7 +462,7 @@ public void onFailure(Exception e) {
 
                 @Override
                 protected void doRun() throws Exception {
-                    beforeSendLatches.get(subRequest).await();
+                    assertTrue(beforeSendLatches.get(subRequest).await(60, TimeUnit.SECONDS));
                     if (client.getLocalNodeId().equals(subRequest.node.getId()) && randomBoolean()) {
                         try {
                             client.executeLocally(TransportTestAction.ACTION, subRequest, latchedListener);