Reintroducing backwards compatibility logic in certain classes

This reverts changes made in opensearch-project#4728 and opensearch-project#4702. These were only made in main and not backported to 2.x
This change also adds unit tests for IndexMetadataGenerations

Signed-off-by: Kartik Ganesh <gkart@amazon.com>

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -1844,11 +1844,15 @@ public static IndexMetadata fromXContent(XContentParser parser) throws IOExcepti
             // Reference:
             // https://github.com/opensearch-project/OpenSearch/blob/4dde0f2a3b445b2fc61dab29c5a2178967f4a3e3/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java#L1620-L1628
             Version legacyVersion = LegacyESVersion.fromId(6050099);
-            if (Assertions.ENABLED && Version.indexCreated(builder.settings).onOrAfter(legacyVersion)) {
+            Version indexCreatedVersion = Version.indexCreated(builder.settings);
+            if (Assertions.ENABLED && indexCreatedVersion.onOrAfter(legacyVersion)) {
                 assert mappingVersion : "mapping version should be present for indices";
                 assert settingsVersion : "settings version should be present for indices";
             }
-            if (Assertions.ENABLED) {
+            // Reference:
+            // https://github.com/opensearch-project/OpenSearch/blob/2e4b27b243d8bd2c515f66cf86c6d1d6a601307f/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java#L1824
+            legacyVersion = LegacyESVersion.fromId(7020099);
+            if (Assertions.ENABLED && indexCreatedVersion.onOrAfter(legacyVersion)) {
                 assert aliasesVersion : "aliases version should be present for indices";
             }
             return builder.build();

server/src/main/java/org/opensearch/repositories/IndexMetaDataGenerations.java

@@ -92,15 +92,21 @@ public String getIndexMetaBlobId(String metaIdentifier) {
     }
 
     /**
-     * Get the blob id by {@link SnapshotId} and {@link IndexId}.
+     * Get the blob id by {@link SnapshotId} and {@link IndexId}. If none is found, we fall back to the value
+     * of {@link SnapshotId#getUUID()} to allow for extended backwards compatibility use-cases with
+     * {@link org.opensearch.LegacyESVersion} versions which used the snapshot UUID as the index metadata blob id.
      *
      * @param snapshotId Snapshot Id
      * @param indexId    Index Id
      * @return blob id for the given index metadata
      */
     public String indexMetaBlobId(SnapshotId snapshotId, IndexId indexId) {
         final String identifier = lookup.getOrDefault(snapshotId, Collections.emptyMap()).get(indexId);
-        return identifiers.get(identifier);
+        if (identifier == null) {
+            return snapshotId.getUUID();
+        } else {
+            return identifiers.get(identifier);
+        }
     }
 
     /**

server/src/test/java/org/opensearch/repositories/IndexMetadataGenerationsTests.java

@@ -0,0 +1,99 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.repositories;
+
+import org.junit.Before;
+import org.opensearch.snapshots.SnapshotId;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class IndexMetadataGenerationsTests extends OpenSearchTestCase {
+
+    private final int MAX_TEST_INDICES = 10;
+    private final String SNAPSHOT = "snapshot";
+    private final String INDEX_PREFIX = "index-";
+    private final String BLOB_ID_PREFIX = "blob-";
+    private IndexMetaDataGenerations indexMetaDataGenerations;
+
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+        final int numIndices = randomIntBetween(1, MAX_TEST_INDICES);
+        Map<IndexId, String> indexMap = createIndexMetadataMap(1, numIndices);
+        Map<String, String> identifierMap = createIdentifierMapFromIndexMetadata(indexMap, BLOB_ID_PREFIX);
+        Map<SnapshotId, Map<IndexId, String>> lookupMap = Collections.singletonMap(new SnapshotId(SNAPSHOT, SNAPSHOT), indexMap);
+        indexMetaDataGenerations = new IndexMetaDataGenerations(lookupMap, identifierMap);
+    }
+
+    public void testEmpty() {
+        assertTrue(IndexMetaDataGenerations.EMPTY.isEmpty());
+        assertNull(IndexMetaDataGenerations.EMPTY.getIndexMetaBlobId("test"));
+    }
+
+    public void testBaseCase() {
+        assertFalse(indexMetaDataGenerations.isEmpty());
+        assertEquals(BLOB_ID_PREFIX + 1, indexMetaDataGenerations.getIndexMetaBlobId(String.valueOf(1)));
+    }
+
+    public void testIndexMetaBlobId() {
+        SnapshotId snapshotId = new SnapshotId(SNAPSHOT, SNAPSHOT);
+        IndexId indexId = new IndexId(INDEX_PREFIX + 1, INDEX_PREFIX + 1);
+        assertEquals(BLOB_ID_PREFIX + 1, indexMetaDataGenerations.indexMetaBlobId(snapshotId, indexId));
+    }
+
+    public void testIndexMetaBlobIdFallback() {
+        SnapshotId snapshotId = new SnapshotId(SNAPSHOT, SNAPSHOT);
+        IndexId indexId = new IndexId("missingIndex", "missingIndex");
+        assertEquals(SNAPSHOT, indexMetaDataGenerations.indexMetaBlobId(snapshotId, indexId));
+
+        final String randomString = randomAlphaOfLength(8);
+        snapshotId = new SnapshotId(randomString, randomString);
+        assertEquals(randomString, indexMetaDataGenerations.indexMetaBlobId(snapshotId, indexId));
+    }
+
+    public void testWithAddedSnapshot() {
+        // Construct a new snapshot
+        SnapshotId newSnapshot = new SnapshotId("newSnapshot", "newSnapshot");
+        final String newIndexMetadataPrefix = "newIndexMetadata-";
+        final String newBlobIdPrefix = "newBlob-";
+        final int numIndices = randomIntBetween(2, MAX_TEST_INDICES);
+        Map<IndexId, String> newLookupMap = createIndexMetadataMap(2, numIndices);
+        Map<String, String> identifierMap = createIdentifierMapFromIndexMetadata(newLookupMap, "newBlob-");
+
+        // Add the snapshot and verify that values have been updated as expected
+        IndexMetaDataGenerations updated = indexMetaDataGenerations.withAddedSnapshot(newSnapshot, newLookupMap, identifierMap);
+        assertEquals(newBlobIdPrefix + 2, updated.getIndexMetaBlobId(String.valueOf(2)));
+        assertEquals(newBlobIdPrefix + 2, updated.indexMetaBlobId(newSnapshot, new IndexId(INDEX_PREFIX + 2, INDEX_PREFIX + 2)));
+        // The first index should remain unchanged
+        assertEquals(BLOB_ID_PREFIX + 1, updated.getIndexMetaBlobId(String.valueOf(1)));
+    }
+
+    public void testWithRemovedSnapshot() {
+        Set<SnapshotId> snapshotToRemove = Collections.singleton(new SnapshotId(SNAPSHOT, SNAPSHOT));
+        assertEquals(IndexMetaDataGenerations.EMPTY, indexMetaDataGenerations.withRemovedSnapshots(snapshotToRemove));
+    }
+
+    private Map<IndexId, String> createIndexMetadataMap(int indexCountLowerBound, int numIndices) {
+        final int indexCountUpperBound = indexCountLowerBound + numIndices;
+        Map<IndexId, String> map = new HashMap<>();
+        for (int i = indexCountLowerBound; i <= indexCountUpperBound; i++) {
+            map.put(new IndexId(INDEX_PREFIX + i, INDEX_PREFIX + i), String.valueOf(i));
+        }
+        return map;
+    }
+
+    private Map<String, String> createIdentifierMapFromIndexMetadata(Map<IndexId, String> indexMetadataMap, String blobIdPrefix) {
+        return indexMetadataMap.values().stream().collect(Collectors.toMap(k -> k, v -> blobIdPrefix + v));
+    }
+}