[ML] Fix and unmute Transform upgrade test (elastic#132995)

(cherry picked from commit 208ae57)

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

Author: davidkyle

muted-tests.yml

@@ -402,9 +402,6 @@ tests:
 - class: org.elasticsearch.packaging.test.ArchiveGenerateInitialCredentialsTests
   method: test50CredentialAutogenerationOnlyOnce
   issue: https://github.com/elastic/elasticsearch/issues/132878
-- class: org.elasticsearch.upgrades.TransformSurvivesUpgradeIT
-  method: testTransformRollingUpgrade
-  issue: https://github.com/elastic/elasticsearch/issues/132892
 - class: org.elasticsearch.xpack.eql.planner.QueryTranslatorTests
   method: testMatchOptimization
   issue: https://github.com/elastic/elasticsearch/issues/132894

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportUpgradeTransformsAction.java

@@ -29,7 +29,6 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.XPackSettings;
-import org.elasticsearch.xpack.core.ml.utils.TransportVersionUtils;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.transform.TransformMetadata;
 import org.elasticsearch.xpack.core.transform.action.UpgradeTransformsAction;
@@ -107,7 +106,7 @@ protected void masterOperation(Task ignoredTask, Request request, ClusterState s
         }
 
         // do not allow in mixed clusters
-        if (TransportVersionUtils.isMinTransportVersionSameAsCurrent(state) == false) {
+        if (state.nodes().isMixedVersionCluster()) {
             listener.onFailure(
                 new ElasticsearchStatusException("Cannot upgrade transforms while cluster upgrade is in progress.", RestStatus.CONFLICT)
             );

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/persistence/IndexBasedTransformConfigManager.java

@@ -295,16 +295,19 @@ public void deleteOldIndices(ActionListener<Boolean> listener) {
 
         // use the transform context as we access system indexes
         try (ThreadContext.StoredContext ctx = client.threadPool().getThreadContext().stashWithOrigin(TRANSFORM_ORIGIN)) {
-            indicesToDelete.addAll(
-                Arrays.asList(
-                    indexNameExpressionResolver.concreteIndexNames(
-                        state,
-                        IndicesOptions.lenientExpandHidden(),
-                        TransformInternalIndexConstants.INDEX_NAME_PATTERN
-                    )
-                )
+            var matchingIndexes = indexNameExpressionResolver.concreteIndices(
+                state,
+                IndicesOptions.lenientExpandHidden(),
+                TransformInternalIndexConstants.INDEX_NAME_PATTERN
             );
 
+            for (var index : matchingIndexes) {
+                var meta = state.getMetadata().indexMetadata(index);
+                if (meta.isSystem() == false) { // ignore system indices as these are automatically managed
+                    indicesToDelete.add(meta.getIndex().getName());
+                }
+            }
+
             indicesToDelete.addAll(
                 Arrays.asList(
                     indexNameExpressionResolver.concreteIndexNames(

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

@@ -73,10 +73,10 @@ public void setUpLogging() throws IOException {
         request.setJsonEntity("""
             {
               "persistent": {
-                "logger.org.elasticsearch.xpack.ml.inference": "TRACE",
+                "logger.org.elasticsearch.xpack.ml.inference": "DEBUG",
                 "logger.org.elasticsearch.xpack.ml.inference.assignments": "DEBUG",
                 "logger.org.elasticsearch.xpack.ml.process": "DEBUG",
-                "logger.org.elasticsearch.xpack.ml.action": "TRACE"
+                "logger.org.elasticsearch.xpack.ml.action": "DEBUG"
               }
             }
             """);

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

@@ -9,6 +9,7 @@
 import org.apache.http.HttpHost;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
@@ -130,11 +131,11 @@ private void createAndStartContinuousTransform() throws Exception {
 
         assertBusy(() -> {
             var stateAndStats = getTransformStats(CONTINUOUS_TRANSFORM_ID);
+            assertThat((Integer) XContentMapValues.extractValue("stats.documents_indexed", stateAndStats), equalTo(ENTITIES.size()));
             assertThat(
-                ((Integer) XContentMapValues.extractValue("stats.documents_indexed", stateAndStats)).longValue(),
-                equalTo(ENTITIES.size())
+                ((Integer) XContentMapValues.extractValue("stats.documents_processed", stateAndStats)).longValue(),
+                equalTo(totalDocsWritten)
             );
-            assertThat((Integer) XContentMapValues.extractValue("stats.documents_processed", stateAndStats), equalTo(totalDocsWritten));
             // Even if we get back to started, we may periodically get set back to `indexing` when triggered.
             // Though short lived due to no changes on the source indices, it could result in flaky test behavior
             assertThat(stateAndStats.get("state"), oneOf("started", "indexing"));
@@ -236,10 +237,12 @@ private void verifyUpgradeFailsIfMixedCluster() {
         if (isOriginalClusterCurrent()) {
             return;
         }
-        final Request upgradeTransformRequest = new Request("POST", getTransformEndpoint() + "_upgrade");
-
-        Exception ex = expectThrows(Exception.class, () -> client().performRequest(upgradeTransformRequest));
-        assertThat(ex.getMessage(), containsString("All nodes must be the same version"));
+        var oldestVersion = Version.fromString(UPGRADE_FROM_VERSION);
+        if (oldestVersion.onOrAfter(Version.V_9_3_0)) {
+            final Request upgradeTransformRequest = new Request("POST", getTransformEndpoint() + "_upgrade");
+            Exception ex = expectThrows(Exception.class, () -> client().performRequest(upgradeTransformRequest));
+            assertThat(ex.getMessage(), containsString("Cannot upgrade transforms while cluster upgrade is in progress"));
+        }
     }
 
     private void verifyUpgrade() throws IOException {