Fix trigger assertion error in watcher tests (#65186)

When we trigger watcher in tests we may fail if scheduler is paused.
That can happen when watcher is reloading due to changes in its shards. This is transient state and retry should fix it.
This change adds assertBusy around triggering code so we will try again.

Closes #65183 #65176 #65117 #65091 #65089 #65086 #65064 #65063

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/actions/TimeThrottleIntegrationTests.java

@@ -31,8 +31,7 @@
 
 public class TimeThrottleIntegrationTests extends AbstractWatcherIntegrationTestCase {
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/65176")
-    public void testTimeThrottle(){
+    public void testTimeThrottle() throws Exception {
         String id = randomAlphaOfLength(20);
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client())
                 .setId(id)
@@ -58,8 +57,7 @@ public void testTimeThrottle(){
         assertTotalHistoryEntries(id, 3);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/65176")
-    public void testTimeThrottleDefaults() {
+    public void testTimeThrottleDefaults() throws Exception {
         String id = randomAlphaOfLength(30);
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client())
                 .setId(id)

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/actions/webhook/WebhookIntegrationTests.java

@@ -71,7 +71,6 @@ public void stopWebservice() throws Exception {
         webServer.close();
     }
 
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/65063")
     public void testWebhook() throws Exception {
         webServer.enqueue(new MockResponse().setResponseCode(200).setBody("body"));
         HttpRequestTemplate.Builder builder = HttpRequestTemplate.builder("localhost", webServer.getPort())

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryActionConditionTests.java

@@ -135,7 +135,6 @@ public void testActionConditionWithHardFailures() throws Exception {
     }
 
     @SuppressWarnings("unchecked")
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/65048")
     public void testActionConditionWithFailures() throws Exception {
         final String id = "testActionConditionWithFailures";
         final ExecutableCondition[] actionConditionsWithFailure = new ExecutableCondition[] {
@@ -180,7 +179,6 @@ public void testActionConditionWithFailures() throws Exception {
     }
 
     @SuppressWarnings("unchecked")
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/65064")
     public void testActionCondition() throws Exception {
         final String id = "testActionCondition";
         final List<ExecutableCondition> actionConditions = new ArrayList<>();
@@ -244,7 +242,7 @@ private List<Object> getActionsFromHit(final Map<String, Object> source) {
      * @param input The input to use for the Watch
      * @param actionConditions The conditions to add to the Watch
      */
-    private void putAndTriggerWatch(final String id, final Input input, final Condition... actionConditions) {
+    private void putAndTriggerWatch(final String id, final Input input, final Condition... actionConditions) throws Exception {
         WatchSourceBuilder source = watchBuilder()
                 .trigger(schedule(interval("5s")))
                 .input(input)

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateEmailMappingsTests.java

@@ -65,7 +65,6 @@ protected Settings nodeSettings(int nodeOrdinal) {
                 .build();
     }
 
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/65052")
     public void testEmailFields() throws Exception {
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client(), "_id").setSource(watchBuilder()
                 .trigger(schedule(interval("5s")))

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateIndexActionMappingsTests.java

@@ -29,7 +29,6 @@
  */
 public class HistoryTemplateIndexActionMappingsTests extends AbstractWatcherIntegrationTestCase {
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/65091")
     public void testIndexActionFields() throws Exception {
         String index = "the-index";
 

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateTimeMappingsTests.java

@@ -32,7 +32,6 @@
  */
 public class HistoryTemplateTimeMappingsTests extends AbstractWatcherIntegrationTestCase {
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/65089")
     public void testTimeFields() throws Exception {
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client(), "_id").setSource(watchBuilder()
                 .trigger(schedule(interval("5s")))

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

@@ -587,23 +587,25 @@ protected static class TimeWarp {
             this.clock = clock;
         }
 
-        public void trigger(String jobName) {
+        public void trigger(String jobName) throws Exception {
             trigger(jobName, 1, null);
         }
 
         public ClockMock clock() {
             return clock;
         }
 
-        public void trigger(String watchId, int times, TimeValue timeValue) {
-            long triggeredCount = schedulers.stream()
-                .filter(scheduler -> scheduler.trigger(watchId, times, timeValue))
-                .count();
-            String msg = String.format(Locale.ROOT, "watch was triggered on [%d] schedulers, expected [1]", triggeredCount);
-            if (triggeredCount > 1) {
-                logger.warn(msg);
-            }
-            assertThat(msg, triggeredCount, greaterThanOrEqualTo(1L));
+        public void trigger(String watchId, int times, TimeValue timeValue) throws Exception {
+            assertBusy(() -> {
+                long triggeredCount = schedulers.stream()
+                    .filter(scheduler -> scheduler.trigger(watchId, times, timeValue))
+                    .count();
+                String msg = String.format(Locale.ROOT, "watch was triggered on [%d] schedulers, expected [1]", triggeredCount);
+                if (triggeredCount > 1) {
+                    logger.warn(msg);
+                }
+                assertThat(msg, triggeredCount, greaterThanOrEqualTo(1L));
+            });
         }
     }
 

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

@@ -108,7 +108,6 @@ protected Map<String, Function<Map<String, Object>, Object>> pluginScripts() {
         }
     }
 
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/65086")
     public void testVars() throws Exception {
         PutWatchResponse putWatchResponse = new PutWatchRequestBuilder(client()).setId(watchId).setSource(watchBuilder()
                 .trigger(schedule(cron("0/1 * * * * ?")))

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/watch/WatchStatusIntegrationTests.java

@@ -33,7 +33,7 @@
 
 public class WatchStatusIntegrationTests extends AbstractWatcherIntegrationTestCase {
 
-    public void testThatStatusGetsUpdated() {
+    public void testThatStatusGetsUpdated() throws Exception {
         new PutWatchRequestBuilder(client(), "_name")
                 .setSource(watchBuilder()
                         .trigger(schedule(interval(5, SECONDS)))