Wait for watcher to be started prior to rolling upgrade tests. (#52186)

Backport: #52139

In the rolling upgrade tests, watcher is manually executed,
in rare scenarios this happens before watcher is started,
resulting in the manual execution to fail.

Relates to #33185

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/UpgradeClusterClientYamlTestSuiteIT.java

@@ -8,6 +8,8 @@
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite;
 import org.apache.lucene.util.TimeUnits;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.test.rest.ESRestTestCase;
@@ -19,6 +21,10 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.equalTo;
 
 
 @TimeoutSuite(millis = 5 * TimeUnits.MINUTE) // to account for slow as hell VMs
@@ -32,6 +38,21 @@ public void waitForTemplates() throws Exception {
         XPackRestTestHelper.waitForTemplates(client(), XPackRestTestConstants.ML_POST_V660_TEMPLATES);
     }
 
+    @Before
+    public void waitForWatcher() throws Exception {
+        // Wait for watcher to be in started state in order to avoid errors due
+        // to manually executing watches prior for watcher to be ready:
+        assertBusy(() -> {
+            Response response = client().performRequest(new Request("GET", "_watcher/stats"));
+            Map<String, Object> responseBody = entityAsMap(response);
+            List<?> stats = (List<?>) responseBody.get("stats");
+            for (Object stat : stats) {
+                Map<?, ?> statAsMap = (Map<?, ?>) stat;
+                assertThat(statAsMap.get("watcher_state"), equalTo("started"));
+            }
+        });
+    }
+
     @Override
     protected boolean preserveIndicesUponCompletion() {
         return true;