Adds util to compare Metadata for remote state (#10661)

* Adds util to compare Metadata for remote state durng restore flow when identifying if 2 clusterUUIDs have same cluster state

Signed-off-by: bansvaru <bansvaru@amazon.com>

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

@@ -922,19 +922,26 @@ public static boolean isGlobalStateEquals(Metadata metadata1, Metadata metadata2
         if (!metadata1.coordinationMetadata.equals(metadata2.coordinationMetadata)) {
             return false;
         }
-        if (!metadata1.persistentSettings.equals(metadata2.persistentSettings)) {
+        if (!metadata1.hashesOfConsistentSettings.equals(metadata2.hashesOfConsistentSettings)) {
             return false;
         }
-        if (!metadata1.hashesOfConsistentSettings.equals(metadata2.hashesOfConsistentSettings)) {
+        if (!metadata1.clusterUUID.equals(metadata2.clusterUUID)) {
             return false;
         }
-        if (!metadata1.templates.equals(metadata2.templates())) {
+        if (metadata1.clusterUUIDCommitted != metadata2.clusterUUIDCommitted) {
             return false;
         }
-        if (!metadata1.clusterUUID.equals(metadata2.clusterUUID)) {
+        return isGlobalResourcesMetadataEquals(metadata1, metadata2);
+    }
+
+    /**
+     * Compares Metadata entities persisted in Remote Store.
+     */
+    public static boolean isGlobalResourcesMetadataEquals(Metadata metadata1, Metadata metadata2) {
+        if (!metadata1.persistentSettings.equals(metadata2.persistentSettings)) {
             return false;
         }
-        if (metadata1.clusterUUIDCommitted != metadata2.clusterUUIDCommitted) {
+        if (!metadata1.templates.equals(metadata2.templates())) {
             return false;
         }
         // Check if any persistent metadata needs to be saved

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -689,10 +689,11 @@ private Map<String, IndexMetadata> getIndexMetadataMap(
      * @return {@link IndexMetadata}
      */
     private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, UploadedIndexMetadata uploadedIndexMetadata) {
+        BlobContainer blobContainer = indexMetadataContainer(clusterName, clusterUUID, uploadedIndexMetadata.getIndexUUID());
         try {
             String[] splitPath = uploadedIndexMetadata.getUploadedFilename().split("/");
             return INDEX_METADATA_FORMAT.read(
-                indexMetadataContainer(clusterName, clusterUUID, uploadedIndexMetadata.getIndexUUID()),
+                blobContainer,
                 splitPath[splitPath.length - 1],
                 blobStoreRepository.getNamedXContentRegistry()
             );
@@ -882,7 +883,8 @@ private Map<String, ClusterMetadataManifest> trimClusterUUIDs(
                 }
             } else {
                 ClusterMetadataManifest previousManifest = trimmedUUIDs.get(currentManifest.getPreviousClusterUUID());
-                if (isMetadataEqual(currentManifest, previousManifest, clusterName)) {
+                if (isMetadataEqual(currentManifest, previousManifest, clusterName)
+                    && isGlobalMetadataEqual(currentManifest, previousManifest, clusterName)) {
                     trimmedUUIDs.remove(clusterUUID);
                 }
             }
@@ -912,6 +914,12 @@ private boolean isMetadataEqual(ClusterMetadataManifest first, ClusterMetadataMa
         return true;
     }
 
+    private boolean isGlobalMetadataEqual(ClusterMetadataManifest first, ClusterMetadataManifest second, String clusterName) {
+        Metadata secondGlobalMetadata = getGlobalMetadata(clusterName, second.getClusterUUID(), second);
+        Metadata firstGlobalMetadata = getGlobalMetadata(clusterName, first.getClusterUUID(), first);
+        return Metadata.isGlobalResourcesMetadataEquals(firstGlobalMetadata, secondGlobalMetadata);
+    }
+
     private boolean isInvalidClusterUUID(ClusterMetadataManifest manifest) {
         return !manifest.isClusterUUIDCommitted();
     }

server/src/test/java/org/opensearch/cluster/metadata/MetadataTests.java

@@ -627,6 +627,39 @@ public void testGlobalStateEqualsCoordinationMetadata() {
         assertFalse(Metadata.isGlobalStateEquals(metadata1, metadata2));
     }
 
+    public void testGlobalResourcesStateEqualsCoordinationMetadata() {
+        CoordinationMetadata coordinationMetadata1 = new CoordinationMetadata(
+            randomNonNegativeLong(),
+            randomVotingConfig(),
+            randomVotingConfig(),
+            randomVotingConfigExclusions()
+        );
+        Metadata metadata1 = Metadata.builder()
+            .coordinationMetadata(coordinationMetadata1)
+            .clusterUUID(randomAlphaOfLength(10))
+            .clusterUUIDCommitted(false)
+            .hashesOfConsistentSettings(Map.of("a", "b"))
+            .persistentSettings(Settings.builder().put(Metadata.SETTING_READ_ONLY_SETTING.getKey(), true).build())
+            .build();
+        CoordinationMetadata coordinationMetadata2 = new CoordinationMetadata(
+            randomNonNegativeLong(),
+            randomVotingConfig(),
+            randomVotingConfig(),
+            randomVotingConfigExclusions()
+        );
+        Metadata metadata2 = Metadata.builder()
+            .coordinationMetadata(coordinationMetadata2)
+            .clusterUUIDCommitted(true)
+            .clusterUUID(randomAlphaOfLength(11))
+            .hashesOfConsistentSettings(Map.of("b", "a"))
+            .persistentSettings(Settings.builder().put(Metadata.SETTING_READ_ONLY_SETTING.getKey(), true).build())
+            .build();
+
+        assertTrue(Metadata.isGlobalStateEquals(metadata1, metadata1));
+        assertFalse(Metadata.isGlobalStateEquals(metadata1, metadata2));
+        assertTrue(Metadata.isGlobalResourcesMetadataEquals(metadata1, metadata2));
+    }
+
     public void testSerializationWithIndexGraveyard() throws IOException {
         final IndexGraveyard graveyard = IndexGraveyardTests.createRandom();
         final Metadata originalMeta = Metadata.builder().indexGraveyard(graveyard).build();