Make SnapshotsInProgress project compatible (#125470)

This PR adds project-id to both SnapshotsInProgress and Snapshot so that
they are aware of projects and ready to handle snapshots from multiple
projects.

Relates: ES-10224

Author: ywangd

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

@@ -0,0 +1,112 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.upgrades;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.rest.ObjectPath;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.elasticsearch.upgrades.SnapshotBasedRecoveryIT.indexDocs;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+
+public class RunningSnapshotIT extends AbstractRollingUpgradeTestCase {
+
+    public RunningSnapshotIT(@Name("upgradedNodes") int upgradedNodes) {
+        super(upgradedNodes);
+    }
+
+    public void testRunningSnapshotCompleteAfterUpgrade() throws Exception {
+        final String indexName = "index";
+        final String repositoryName = "repo";
+        final String snapshotName = "snapshot";
+        final var nodeIds = getNodesInfo(client()).keySet();
+
+        if (isOldCluster()) {
+            registerRepository(repositoryName, "fs", randomBoolean(), Settings.builder().put("location", "backup").build());
+            // create an index to have one shard per node
+            createIndex(indexName, indexSettings(3, 0).put("index.routing.allocation.total_shards_per_node", 1).build());
+            ensureGreen(indexName);
+            if (randomBoolean()) {
+                indexDocs(indexName, between(10, 50));
+            }
+            flush(indexName, true);
+            // Signal shutdown to prevent snapshot from being completed
+            putShutdownMetadata(nodeIds);
+            createSnapshot(repositoryName, snapshotName, false);
+            assertRunningSnapshot(repositoryName, snapshotName);
+        } else {
+            if (isUpgradedCluster()) {
+                deleteShutdownMetadata(nodeIds);
+                assertNoShutdownMetadata(nodeIds);
+                ensureGreen(indexName);
+                assertBusy(() -> assertCompletedSnapshot(repositoryName, snapshotName));
+            } else {
+                assertRunningSnapshot(repositoryName, snapshotName);
+            }
+        }
+    }
+
+    private void putShutdownMetadata(Collection<String> nodeIds) throws IOException {
+        for (String nodeId : nodeIds) {
+            final Request putShutdownRequest = new Request("PUT", "/_nodes/" + nodeId + "/shutdown");
+            putShutdownRequest.setJsonEntity("""
+                {
+                  "type": "remove",
+                  "reason": "test"
+                }""");
+            client().performRequest(putShutdownRequest);
+        }
+    }
+
+    private void deleteShutdownMetadata(Collection<String> nodeIds) throws IOException {
+        for (String nodeId : nodeIds) {
+            final Request request = new Request("DELETE", "/_nodes/" + nodeId + "/shutdown");
+            client().performRequest(request);
+        }
+    }
+
+    private void assertNoShutdownMetadata(Collection<String> nodeIds) throws IOException {
+        final ObjectPath responsePath = assertOKAndCreateObjectPath(
+            client().performRequest(new Request("GET", "/_nodes/" + Strings.collectionToCommaDelimitedString(nodeIds) + "/shutdown"))
+        );
+        assertThat(responsePath.evaluate("nodes"), empty());
+    }
+
+    private void assertRunningSnapshot(String repositoryName, String snapshotName) throws IOException {
+        final Request request = new Request("GET", "/_snapshot/" + repositoryName + "/_current");
+        final ObjectPath responsePath = assertOKAndCreateObjectPath(client().performRequest(request));
+        assertThat(responsePath.evaluate("total"), equalTo(1));
+        assertThat(responsePath.evaluate("snapshots.0.snapshot"), equalTo(snapshotName));
+    }
+
+    private void assertCompletedSnapshot(String repositoryName, String snapshotName) throws IOException {
+        {
+            final Request request = new Request("GET", "/_snapshot/" + repositoryName + "/_current");
+            final ObjectPath responsePath = assertOKAndCreateObjectPath(client().performRequest(request));
+            assertThat(responsePath.evaluate("total"), equalTo(0));
+        }
+
+        {
+            final Request request = new Request("GET", "/_snapshot/" + repositoryName + "/" + snapshotName);
+            final ObjectPath responsePath = assertOKAndCreateObjectPath(client().performRequest(request));
+            assertThat(responsePath.evaluate("total"), equalTo(1));
+            assertThat(responsePath.evaluate("snapshots.0.snapshot"), equalTo(snapshotName));
+            assertThat(responsePath.evaluate("snapshots.0.state"), not(equalTo("IN_PROGRESS")));
+        }
+    }
+}

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

@@ -233,7 +233,7 @@ private static Map<String, Object> search(String index, QueryBuilder query) thro
         return responseAsMap;
     }
 
-    private void indexDocs(String indexName, int numDocs) throws IOException {
+    static void indexDocs(String indexName, int numDocs) throws IOException {
         final StringBuilder bulkBody = new StringBuilder();
         for (int i = 0; i < numDocs; i++) {
             bulkBody.append("{\"index\":{\"_id\":\"").append(i).append("\"}}\n");

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -205,6 +205,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion RERANK_COMMON_OPTIONS_ADDED = def(9_037_0_00);
     public static final TransportVersion ESQL_REPORT_ORIGINAL_TYPES = def(9_038_00_0);
     public static final TransportVersion RESCORE_VECTOR_ALLOW_ZERO = def(9_039_0_00);
+    public static final TransportVersion PROJECT_ID_IN_SNAPSHOT = def(9_040_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java

@@ -146,6 +146,29 @@ public static <K, T, M extends Map<K, T>> boolean hasKey(MapDiff<K, T, M> diff,
         return false;
     }
 
+    /**
+     * Create a new JDK map backed MapDiff by transforming the keys with the provided keyFunction.
+     * @param diff Original MapDiff to transform
+     * @param keyFunction Function to transform the key
+     * @param keySerializer Serializer for the new key
+     */
+    public static <K1, K2, T extends Diffable<T>, M1 extends Map<K1, T>> MapDiff<K2, T, Map<K2, T>> jdkMapDiffWithUpdatedKeys(
+        MapDiff<K1, T, M1> diff,
+        Function<K1, K2> keyFunction,
+        KeySerializer<K2> keySerializer
+    ) {
+        final List<K2> deletes = diff.getDeletes().stream().map(keyFunction).toList();
+        final List<Map.Entry<K2, Diff<T>>> diffs = diff.getDiffs()
+            .stream()
+            .map(entry -> Map.entry(keyFunction.apply(entry.getKey()), entry.getValue()))
+            .toList();
+        final List<Map.Entry<K2, T>> upserts = diff.getUpserts()
+            .stream()
+            .map(entry -> Map.entry(keyFunction.apply(entry.getKey()), entry.getValue()))
+            .toList();
+        return new MapDiff<>(keySerializer, DiffableValueSerializer.getWriteOnlyInstance(), deletes, diffs, upserts, JdkMapBuilder::new);
+    }
+
     /**
      * Creates a MapDiff that applies a single entry diff to a map
      */