Watcher fix ActivateWatchTests (#82157)

This commit attempts to fix a set of Watcher tests that can
fail due to unexpected execution of Watches after Watcher has been stopped.

The theory here is that a Watch can be queued but not fully executed
then Watcher is shutdown, the test does some clean up, then the
queued Watch finishes execution and causes some some additional cleanup
to fail.

The change here ensures that when Watcher is stopped from AbstractWatcherIntegrationTestCase
that it will also wait until there are no more current Watches executing.

closes #66495

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/test/AbstractWatcherIntegrationTestCase.java

@@ -567,7 +567,8 @@ public abstract class AbstractWatcherIntegrationTestCase extends ESIntegTestCase
 
     protected void stopWatcher() throws Exception {
         assertBusy(() -> {
-            WatcherStatsResponse watcherStatsResponse = new WatcherStatsRequestBuilder(client()).get();
+
+            WatcherStatsResponse watcherStatsResponse = new WatcherStatsRequestBuilder(client()).setIncludeCurrentWatches(true).get();
             assertThat(watcherStatsResponse.hasFailures(), is(false));
             List<Tuple<String, WatcherState>> currentStatesFromStatsRequest = watcherStatsResponse.getNodes()
                 .stream()
@@ -580,7 +581,8 @@ public abstract class AbstractWatcherIntegrationTestCase extends ESIntegTestCase
                 .collect(Collectors.toList());
             List<WatcherState> states = currentStatesFromStatsRequest.stream().map(Tuple::v2).collect(Collectors.toList());
 
-            logger.info("waiting to stop watcher, current states {}", currentStatesFromStatsRequest);
+            long currentWatches = watcherStatsResponse.getNodes().stream().mapToLong(n -> n.getSnapshots().size()).sum();
+            logger.info("waiting to stop watcher, current states {}, current watches [{}]", currentStatesFromStatsRequest, currentWatches);
 
             boolean isAllStateStarted = states.stream().allMatch(w -> w == WatcherState.STARTED);
             if (isAllStateStarted) {
@@ -594,7 +596,7 @@ public abstract class AbstractWatcherIntegrationTestCase extends ESIntegTestCase
             }
 
             boolean isAllStateStopped = states.stream().allMatch(w -> w == WatcherState.STOPPED);
-            if (isAllStateStopped) {
+            if (isAllStateStopped && currentWatches == 0) {
                 return;
             }
 

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/transport/action/activate/ActivateWatchTests.java

@@ -50,7 +50,6 @@ public class ActivateWatchTests extends AbstractWatcherIntegrationTestCase {
         return false;
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/66495")
     public void testDeactivateAndActivate() throws Exception {
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client()).setId("_id")
             .setSource(
@@ -107,7 +106,6 @@ public class ActivateWatchTests extends AbstractWatcherIntegrationTestCase {
         });
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/66495")
     public void testLoadWatchWithoutAState() throws Exception {
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client()).setId("_id")
             .setSource(