[Test] Allow allocation in mixed cluster

The RunningSnapshotIT upgrade test adds shutdown marker to all nodes and
removed them once all nodes are upgraded. If an index gets created in a
mixed cluster, for example by ILM or deprecation messages, the index
cannot be allocated because all nodes are shutting down. Since the
cluster ready check between node upgrades expects a yellow cluster, the
unassigned index prevents the ready check to succeed and eventually
timeout. This PR fixes it by removing shutdown marker for the 1st
upgrade node to allow it hosting new indices.

Resolves: #129644
Resolves: #129645
Resolves: #129646

Author: ywangd

muted-tests.yml

@@ -571,15 +571,6 @@ tests:
 - class: org.elasticsearch.server.cli.MachineDependentHeapTests
   method: testMlOnlyOptions
   issue: https://github.com/elastic/elasticsearch/issues/129236
-- class: org.elasticsearch.upgrades.RunningSnapshotIT
-  method: testRunningSnapshotCompleteAfterUpgrade {upgradedNodes=1}
-  issue: https://github.com/elastic/elasticsearch/issues/129644
-- class: org.elasticsearch.upgrades.RunningSnapshotIT
-  method: testRunningSnapshotCompleteAfterUpgrade {upgradedNodes=2}
-  issue: https://github.com/elastic/elasticsearch/issues/129645
-- class: org.elasticsearch.upgrades.RunningSnapshotIT
-  method: testRunningSnapshotCompleteAfterUpgrade {upgradedNodes=3}
-  issue: https://github.com/elastic/elasticsearch/issues/129646
 - class: org.elasticsearch.test.apmintegration.TracesApmIT
   method: testApmIntegration
   issue: https://github.com/elastic/elasticsearch/issues/129651

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/RunningSnapshotIT.java

@@ -18,7 +18,10 @@
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
 
+import static org.elasticsearch.client.RestClient.IGNORE_RESPONSE_CODES_PARAM;
 import static org.elasticsearch.upgrades.SnapshotBasedRecoveryIT.indexDocs;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
@@ -34,7 +37,10 @@ public void testRunningSnapshotCompleteAfterUpgrade() throws Exception {
         final String indexName = "index";
         final String repositoryName = "repo";
         final String snapshotName = "snapshot";
-        final var nodeIds = getNodesInfo(client()).keySet();
+        final Map<String, Map<?, ?>> nodesInfo = getNodesInfo(client());
+        final var nodeIdToBuildHashes = nodesInfo.entrySet()
+            .stream()
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry -> entry.getValue().get("build_hash").toString()));
 
         if (isOldCluster()) {
             registerRepository(repositoryName, "fs", randomBoolean(), Settings.builder().put("location", "backup").build());
@@ -46,16 +52,27 @@ public void testRunningSnapshotCompleteAfterUpgrade() throws Exception {
             }
             flush(indexName, true);
             // Signal shutdown to prevent snapshot from being completed
-            putShutdownMetadata(nodeIds);
+            putShutdownMetadata(nodeIdToBuildHashes.keySet());
             createSnapshot(repositoryName, snapshotName, false);
             assertRunningSnapshot(repositoryName, snapshotName);
         } else {
             if (isUpgradedCluster()) {
-                deleteShutdownMetadata(nodeIds);
-                assertNoShutdownMetadata(nodeIds);
+                deleteShutdownMetadata(nodeIdToBuildHashes.keySet());
+                assertNoShutdownMetadata(nodeIdToBuildHashes.keySet());
                 ensureGreen(indexName);
                 assertBusy(() -> assertCompletedSnapshot(repositoryName, snapshotName));
             } else {
+                final var buildHashToNodeIds = nodeIdToBuildHashes.entrySet()
+                    .stream()
+                    .collect(Collectors.groupingBy(Map.Entry::getValue, Collectors.mapping(Map.Entry::getKey, Collectors.toSet())));
+                if (isFirstMixedCluster()) {
+                    final var upgradedNodeIds = buildHashToNodeIds.values()
+                        .stream()
+                        .filter(strings -> strings.size() == 1)
+                        .findFirst()
+                        .orElseThrow(() -> new AssertionError("expect one upgraded node, but got " + buildHashToNodeIds));
+                    deleteShutdownMetadata(upgradedNodeIds);
+                }
                 assertRunningSnapshot(repositoryName, snapshotName);
             }
         }
@@ -76,6 +93,7 @@ private void putShutdownMetadata(Collection<String> nodeIds) throws IOException
     private void deleteShutdownMetadata(Collection<String> nodeIds) throws IOException {
         for (String nodeId : nodeIds) {
             final Request request = new Request("DELETE", "/_nodes/" + nodeId + "/shutdown");
+            request.addParameter(IGNORE_RESPONSE_CODES_PARAM, "404");
             client().performRequest(request);
         }
     }