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

(cherry picked from commit 6257be7)

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

Author: DonalEvans

muted-tests.yml

@@ -290,9 +290,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

@@ -47,7 +47,6 @@
 import org.elasticsearch.xpack.core.ml.job.results.Result;
 import org.elasticsearch.xpack.core.ml.job.snapshot.upgrade.SnapshotUpgradeTaskParams;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
-import org.elasticsearch.xpack.core.ml.utils.TransportVersionUtils;
 import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.ml.job.persistence.JobConfigProvider;
 import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider;
@@ -110,7 +109,7 @@ protected void masterOperation(Task task, Request request, ClusterState state, A
             return;
         }
 
-        if (TransportVersionUtils.isMinTransportVersionSameAsCurrent(state) == 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).after(Version.V_9_2_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);