Fix CCS cancellation test (#117790) (#117799)

We should have checked that all drivers were canceled, not cancellable
(which is always true), before unblocking the compute tasks.

Closes #117568

Author: dnhatn

muted-tests.yml

@@ -408,6 +408,3 @@ tests:
 - class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
   method: test {scoring.QstrWithFieldAndScoringSortedEval}
   issue: https://github.com/elastic/elasticsearch/issues/117751
-- class: org.elasticsearch.search.ccs.CrossClusterIT
-  method: testCancel
-  issue: https://github.com/elastic/elasticsearch/issues/108061

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersCancellationIT.java

@@ -179,19 +179,22 @@ public void testCancel() throws Exception {
         });
         var cancelRequest = new CancelTasksRequest().setTargetTaskId(rootTasks.get(0).taskId()).setReason("proxy timeout");
         client().execute(TransportCancelTasksAction.TYPE, cancelRequest);
-        assertBusy(() -> {
-            List<TaskInfo> drivers = client(REMOTE_CLUSTER).admin()
-                .cluster()
-                .prepareListTasks()
-                .setActions(DriverTaskRunner.ACTION_NAME)
-                .get()
-                .getTasks();
-            assertThat(drivers.size(), greaterThanOrEqualTo(1));
-            for (TaskInfo driver : drivers) {
-                assertTrue(driver.cancellable());
-            }
-        });
-        PauseFieldPlugin.allowEmitting.countDown();
+        try {
+            assertBusy(() -> {
+                List<TaskInfo> drivers = client(REMOTE_CLUSTER).admin()
+                    .cluster()
+                    .prepareListTasks()
+                    .setActions(DriverTaskRunner.ACTION_NAME)
+                    .get()
+                    .getTasks();
+                assertThat(drivers.size(), greaterThanOrEqualTo(1));
+                for (TaskInfo driver : drivers) {
+                    assertTrue(driver.cancelled());
+                }
+            });
+        } finally {
+            PauseFieldPlugin.allowEmitting.countDown();
+        }
         Exception error = expectThrows(Exception.class, requestFuture::actionGet);
         assertThat(error.getMessage(), containsString("proxy timeout"));
     }