QA: Fixup rolling restart tests

Fix the skip message in the token tests to explain why the test can only
be run once in the mixed version cluster.

Reenable the watcher tests by writing a condition that will hold true.

Author: nik9000

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

@@ -72,8 +72,8 @@ public void testTokenWorksInMixedOrUpgradedCluster() throws Exception {
     public void testMixedCluster() throws Exception {
         assumeTrue("this test should only run against the mixed cluster", CLUSTER_TYPE == ClusterType.MIXED);
         assumeTrue("the master must be on the latest version before we can write", isMasterOnLatestVersion());
-        assumeFalse("Can't be run twice in mixed clusters for unknown reasons so we skip the first attempt",
-                false == Booleans.parseBoolean(System.getProperty("tests.first_round")));
+        assumeFalse("can't be run twice because it invalidates a token so we skip the first attempt",
+                Booleans.parseBoolean(System.getProperty("tests.first_round")));
 
         Response getResponse = client().performRequest("GET", "token_backwards_compatibility_it/doc/old_cluster_token2");
         assertOK(getResponse);

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

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.xpack.watcher.actions.ActionBuilders.loggingAction;
@@ -177,14 +178,14 @@ public void testWatcherStats() throws Exception {
     }
 
     public void testWatcherRestart() throws Exception {
-        assumeFalse("Seems to be broken in mixed clusters. Skipping while I debug.", CLUSTER_TYPE == ClusterType.MIXED);
         executeUpgradeIfNeeded();
 
         executeAgainstRandomNode(client -> assertOK(client.performRequest("POST", "/_xpack/watcher/_stop")));
         ensureWatcherStopped();
 
         executeAgainstRandomNode(client -> assertOK(client.performRequest("POST", "/_xpack/watcher/_start")));
-        ensureWatcherStarted();
+        // Watcher should be started on at least the nodes with the new version.
+        ensureWatcherStartedOnModernNodes();
     }
 
     public void testWatchCrudApis() throws Exception {
@@ -314,6 +315,30 @@ private void ensureWatcherStarted() throws Exception {
         }));
     }
 
+    private void ensureWatcherStartedOnModernNodes() throws Exception {
+        if (nodes.getMaster().getVersion().before(Version.V_6_0_0)) {
+            /*
+             * Versions before 6.0 ran watcher on the master node and the
+             * logic in ensureWatcherStarted is fine.
+             */
+            ensureWatcherStarted();
+            return;
+        }
+        executeAgainstMasterNode(client -> assertBusy(() -> {
+            Map<?, ?> responseBody = entityAsMap(client.performRequest("GET", "_xpack/watcher/stats"));
+            logger.info("ensureWatcherStartedOnModernNodes(), stats response [{}]", responseBody);
+            Map<?, ?> stats = ((List<?>) responseBody.get("stats")).stream()
+                .map(o -> (Map<?, ?>) o)
+                .collect(Collectors.toMap(m -> m.get("node_id"), Function.identity()));
+            assertNotNull("no stats yet", stats);
+            for (Node node : nodes.getNewNodes()) {
+                Map<?, ?> nodeStats = (Map<?, ?>) stats.get(node.getId());
+                assertEquals("modern node [" + node.getId() + "] is not started",
+                        nodeStats.get("watcher_state"), "started");
+            }
+        }));
+    }
+
     private Nodes buildNodeAndVersions() throws IOException {
         Response response = client().performRequest("GET", "_nodes");
         ObjectPath objectPath = ObjectPath.createFromResponse(response);