Fix snapshot upgrade test (elastic#136433)

Using transport version to determine if a cluster is mixed version is
sometimes unreliable, so use DiscoveryNodes versions instead

Closes elastic#98560

Author: DonalEvans

muted-tests.yml

@@ -281,9 +281,6 @@ tests:
 - class: org.elasticsearch.xpack.profiling.action.GetStatusActionIT
   method: testWaitsUntilResourcesAreCreated
   issue: https://github.com/elastic/elasticsearch/issues/129486
-- class: org.elasticsearch.upgrades.MlJobSnapshotUpgradeIT
-  method: testSnapshotUpgrader
-  issue: https://github.com/elastic/elasticsearch/issues/98560
 - class: org.elasticsearch.search.query.VectorIT
   method: testFilteredQueryStrategy
   issue: https://github.com/elastic/elasticsearch/issues/129517

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpgradeJobModelSnapshotAction.java

@@ -11,7 +11,6 @@
 import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.ResourceNotFoundException;
-import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@@ -110,7 +109,7 @@ protected void masterOperation(Task task, Request request, ClusterState state, A
             return;
         }
 
-        if (state.getMinTransportVersion().equals(TransportVersion.current()) == false) {
+        if (state.nodes().isMixedVersionCluster()) {
             listener.onFailure(
                 ExceptionsHelper.conflictStatusException(
                     "Cannot upgrade job [{}] snapshot [{}] while cluster upgrade is in progress.",

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

@@ -7,6 +7,7 @@
 package org.elasticsearch.upgrades;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
@@ -74,6 +75,10 @@ public void testSnapshotUpgrader() throws Exception {
             case OLD -> createJobAndSnapshots();
             case MIXED -> {
                 assumeTrue("We should only test if old cluster is before new cluster", isOriginalClusterCurrent() == false);
+                assumeTrue(
+                    "Older versions could not always reliably determine if we were in a mixed cluster state",
+                    Version.fromString(UPGRADE_FROM_VERSION).onOrAfter(Version.V_9_3_0)
+                );
                 ensureHealth((request -> {
                     request.addParameter("timeout", "70s");
                     request.addParameter("wait_for_nodes", "3");
@@ -97,13 +102,6 @@ public void testSnapshotUpgrader() throws Exception {
 
     @SuppressWarnings("unchecked")
     private void testSnapshotUpgradeFailsOnMixedCluster() throws Exception {
-        // TODO the mixed cluster assertions sometimes fail because the code that
-        // detects the mixed cluster relies on the transport versions being different.
-        // This assumption does not hold immediately after a version bump and new
-        // branch being cut as the new branch will have the same transport version
-        // See https://github.com/elastic/elasticsearch/issues/98560
-
-        assumeTrue("The mixed cluster is not always detected correctly, see https://github.com/elastic/elasticsearch/issues/98560", false);
         Map<String, Object> jobs = entityAsMap(getJob(JOB_ID));
 
         String currentSnapshot = ((List<String>) XContentMapValues.extractValue("jobs.model_snapshot_id", jobs)).get(0);