[Backport 2.x] Fix stale cluster state custom file deletion

CHANGELOG.md

@@ -47,6 +47,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix rollover alias supports restored searchable snapshot index([#16483](https://github.com/opensearch-project/OpenSearch/pull/16483))
 - Fix permissions error on scripted query against remote snapshot ([#16544](https://github.com/opensearch-project/OpenSearch/pull/16544))
 - Fix `doc_values` only (`index:false`) IP field searching for masks ([#16628](https://github.com/opensearch-project/OpenSearch/pull/16628))
+- Fix stale cluster state custom file deletion ([#16670](https://github.com/opensearch-project/OpenSearch/pull/16670))
 
 ### Security
 

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

@@ -306,7 +306,7 @@ void deleteClusterMetadata(
                     staleEphemeralAttributePaths.add(clusterMetadataManifest.getHashesOfConsistentSettings().getUploadedFilename());
                 }
                 if (clusterMetadataManifest.getClusterStateCustomMap() != null) {
-                    clusterMetadataManifest.getCustomMetadataMap()
+                    clusterMetadataManifest.getClusterStateCustomMap()
                         .values()
                         .stream()
                         .filter(u -> !filesToKeep.contains(u.getUploadedFilename()))

server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerTests.java

@@ -42,6 +42,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -200,6 +201,7 @@ public void testDeleteClusterMetadata() throws IOException {
             .nodeId("nodeA")
             .opensearchVersion(VersionUtils.randomOpenSearchVersion(random()))
             .previousClusterUUID(ClusterState.UNKNOWN_UUID)
+            .clusterStateCustomMetadataMap(Map.of("snapshots", new UploadedMetadataAttribute("snapshots", "snapshot_file1")))
             .committed(true)
             .build();
         ClusterMetadataManifest manifest2 = ClusterMetadataManifest.builder(manifest1)
@@ -209,10 +211,12 @@ public void testDeleteClusterMetadata() throws IOException {
             .coordinationMetadata(coordinationMetadata)
             .templatesMetadata(templateMetadata)
             .settingMetadata(settingMetadata)
+            .clusterStateCustomMetadataMap(Map.of("restore", new UploadedMetadataAttribute("restore", "restore_file1")))
             .build();
         ClusterMetadataManifest manifest3 = ClusterMetadataManifest.builder(manifest2)
             .indices(List.of(index1UpdatedMetadata, index2Metadata))
             .settingMetadata(settingMetadataUpdated)
+            .clusterStateCustomMetadataMap(Map.of())
             .build();
 
         UploadedIndexMetadata index3Metadata = new UploadedIndexMetadata("index3", "indexUUID3", "index_metadata3__2");
@@ -286,6 +290,7 @@ public void testDeleteClusterMetadata() throws IOException {
             )
         );
         verify(container).deleteBlobsIgnoringIfNotExists(List.of(getFormattedIndexFileName(index1Metadata.getUploadedFilePath())));
+        verify(container).deleteBlobsIgnoringIfNotExists(List.of("restore_file1", "snapshot_file1"));
         Set<String> staleManifest = new HashSet<>();
         inactiveBlobs.forEach(
             blob -> staleManifest.add(