Add opensearch version info while deserialization

CHANGELOG.md

@@ -102,6 +102,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Workload Management] Fixing Create/Update QueryGroup TransportActions to execute from non-cluster manager nodes ([16422](https://github.com/opensearch-project/OpenSearch/pull/16422))
 - Fix flaky test in `testApproximateRangeWithSizeOverDefault` by adjusting totalHits assertion logic ([#16434](https://github.com/opensearch-project/OpenSearch/pull/16434#pullrequestreview-2386999409))
 - Revert changes to upload remote state manifest using minimum codec version([#16403](https://github.com/opensearch-project/OpenSearch/pull/16403))
+- Use OpenSearch version to deserialize remote custom metadata([#16494](https://github.com/opensearch-project/OpenSearch/pull/16494))
 
 ### Security
 

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

@@ -1262,7 +1262,8 @@ ClusterState readClusterStateInParallel(
                     entry.getKey(),
                     clusterUUID,
                     blobStoreRepository.getCompressor(),
-                    namedWriteableRegistry
+                    namedWriteableRegistry,
+                    manifest.getOpensearchVersion()
                 ),
                 listener
             );

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

@@ -242,7 +242,8 @@
                             key,
                             clusterUUID,
                             compressor,
-                            namedWriteableRegistry
+                            namedWriteableRegistry,
+                            clusterMetadataManifest.getOpensearchVersion()
                         );
                         builder.putCustom(key, (Custom) getStore(remoteCustomMetadata).read(remoteCustomMetadata));
                     } catch (IOException e) {

server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java

@@ -8,6 +8,7 @@
 
 package org.opensearch.gateway.remote.model;
 
+import org.opensearch.Version;
 import org.opensearch.cluster.metadata.Metadata.Custom;
 import org.opensearch.common.io.Streams;
 import org.opensearch.common.remote.AbstractClusterMetadataWriteableBlobEntity;
@@ -67,16 +68,17 @@ public RemoteCustomMetadata(
         final String customType,
         final String clusterUUID,
         final Compressor compressor,
-        final NamedWriteableRegistry namedWriteableRegistry
+        final NamedWriteableRegistry namedWriteableRegistry,
+        final Version version
     ) {
         super(clusterUUID, compressor, null);
         this.blobName = blobName;
         this.customType = customType;
         this.namedWriteableRegistry = namedWriteableRegistry;
-        this.customBlobStoreFormat = new ChecksumWritableBlobStoreFormat<>(
-            "custom",
-            is -> readFrom(is, namedWriteableRegistry, customType)
-        );
+        this.customBlobStoreFormat = new ChecksumWritableBlobStoreFormat<>("custom", is -> {
+            is.setVersion(version);
+            return readFrom(is, namedWriteableRegistry, customType);
+        });
     }
 
     @Override

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

@@ -8,6 +8,7 @@
 
 package org.opensearch.gateway.remote;
 
+import org.opensearch.Version;
 import org.opensearch.action.LatchedActionListener;
 import org.opensearch.cluster.ClusterModule;
 import org.opensearch.cluster.ClusterName;
@@ -480,14 +481,21 @@ public void testGetAsyncWriteRunnable_TemplatesMetadata() throws Exception {
     }
 
     public void testGetAsyncReadRunnable_CustomMetadata() throws Exception {
+        for (Version version : List.of(Version.CURRENT, Version.V_2_15_0, Version.V_2_13_0)) {
+            verifyCustomMetadataReadForVersion(version);
+        }
+    }
+
+    private void verifyCustomMetadataReadForVersion(Version version) throws Exception {
         Metadata.Custom customMetadata = getCustomMetadata();
         String fileName = randomAlphaOfLength(10);
         RemoteCustomMetadata customMetadataForDownload = new RemoteCustomMetadata(
             fileName,
             IndexGraveyard.TYPE,
             CLUSTER_UUID,
             compressor,
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            version
         );
         when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn(
             customMetadataForDownload.customBlobStoreFormat.serialize(customMetadata, fileName, compressor).streamInput()
@@ -695,4 +703,5 @@ public void testGetUpdatedCustoms() {
         assertThat(customsDiff.getUpserts(), is(expectedUpserts));
         assertThat(customsDiff.getDeletes(), is(List.of(CustomMetadata1.TYPE)));
     }
+
 }

server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java

@@ -106,7 +106,8 @@ public void testClusterUUID() {
             "test-custom",
             clusterUUID,
             compressor,
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            Version.CURRENT
         );
         assertThat(remoteObjectForDownload.clusterUUID(), is(clusterUUID));
     }
@@ -128,7 +129,8 @@ public void testFullBlobName() {
             "test-custom",
             clusterUUID,
             compressor,
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            Version.CURRENT
         );
         assertThat(remoteObjectForDownload.getFullBlobName(), is(TEST_BLOB_NAME));
     }
@@ -150,7 +152,8 @@ public void testBlobFileName() {
             "test-custom",
             clusterUUID,
             compressor,
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            Version.CURRENT
         );
         assertThat(remoteObjectForDownload.getBlobFileName(), is(TEST_BLOB_FILE_NAME));
     }
@@ -162,7 +165,8 @@ public void testBlobPathTokens() {
             "test-custom",
             clusterUUID,
             compressor,
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            Version.CURRENT
         );
         assertThat(remoteObjectForDownload.getBlobPathTokens(), is(new String[] { "user", "local", "opensearch", "customMetadata" }));
     }