From 4ea63a5bbd4386e70ef89a5c947dc73f2e266ce3 Mon Sep 17 00:00:00 2001
From: Sooraj Sinha <soosinha@amazon.com>
Date: Mon, 21 Oct 2024 17:49:31 +0530
Subject: [PATCH] Serialize using codec version

Signed-off-by: Sooraj Sinha <soosinha@amazon.com>
---
 .../gateway/remote/ClusterMetadataManifest.java  | 14 ++++++++++++--
 .../gateway/remote/RemoteManifestManager.java    |  1 +
 .../model/RemoteClusterMetadataManifest.java     | 16 +++++++++++++---
 3 files changed, 26 insertions(+), 5 deletions(-)

diff --git a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java
index 405e5cd784196..c7c62c10026eb 100644
--- a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java
+++ b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java
@@ -1216,12 +1216,18 @@ private static String componentPrefix(Object[] fields) {
 
         private static final ConstructingObjectParser<UploadedIndexMetadata, Void> PARSER_V0 = new ConstructingObjectParser<>(
             "uploaded_index_metadata",
-            fields -> new UploadedIndexMetadata(indexName(fields), indexUUID(fields), uploadedFilename(fields))
+            fields -> new UploadedIndexMetadata(indexName(fields), indexUUID(fields), uploadedFilename(fields), CODEC_V0)
         );
 
         private static final ConstructingObjectParser<UploadedIndexMetadata, Void> PARSER_V2 = new ConstructingObjectParser<>(
             "uploaded_index_metadata",
-            fields -> new UploadedIndexMetadata(indexName(fields), indexUUID(fields), uploadedFilename(fields), componentPrefix(fields))
+            fields -> new UploadedIndexMetadata(
+                indexName(fields),
+                indexUUID(fields),
+                uploadedFilename(fields),
+                componentPrefix(fields),
+                CODEC_V2
+            )
         );
 
         private static final ConstructingObjectParser<UploadedIndexMetadata, Void> CURRENT_PARSER = PARSER_V2;
@@ -1306,6 +1312,10 @@ public String getComponentPrefix() {
             return componentPrefix;
         }
 
+        protected void setCodecVersion(long codecVersion) {
+            this.codecVersion = codecVersion;
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.field(INDEX_NAME_FIELD.getPreferredName(), getIndexName())
diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java
index b243269fe323e..1047722b6d883 100644
--- a/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java
+++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java
@@ -105,6 +105,7 @@ RemoteClusterStateManifestInfo uploadManifest(
     ) {
         synchronized (this) {
             ClusterMetadataManifest.Builder manifestBuilder = ClusterMetadataManifest.builder();
+            uploadedMetadataResult.uploadedIndexMetadata.forEach(md -> md.setCodecVersion(codecVersion));
             manifestBuilder.clusterTerm(clusterState.term())
                 .stateVersion(clusterState.getVersion())
                 .clusterUUID(clusterState.metadata().clusterUUID())
diff --git a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterMetadataManifest.java b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterMetadataManifest.java
index 999beaa4e865d..5c7fb61d90684 100644
--- a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterMetadataManifest.java
+++ b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterMetadataManifest.java
@@ -123,7 +123,8 @@ public UploadedMetadata getUploadedMetadata() {
 
     @Override
     public InputStream serialize() throws IOException {
-        return CLUSTER_METADATA_MANIFEST_FORMAT.serialize(
+        ChecksumBlobStoreFormat<ClusterMetadataManifest> blobStoreFormat = getClusterMetadataManifestBlobStoreFormatForUpload();
+        return blobStoreFormat.serialize(
             clusterMetadataManifest,
             generateBlobFileName(),
             getCompressor(),
@@ -133,7 +134,7 @@ public InputStream serialize() throws IOException {
 
     @Override
     public ClusterMetadataManifest deserialize(final InputStream inputStream) throws IOException {
-        ChecksumBlobStoreFormat<ClusterMetadataManifest> blobStoreFormat = getClusterMetadataManifestBlobStoreFormat();
+        ChecksumBlobStoreFormat<ClusterMetadataManifest> blobStoreFormat = getClusterMetadataManifestBlobStoreFormatForDownload();
         return blobStoreFormat.deserialize(blobName, getNamedXContentRegistry(), Streams.readFully(inputStream));
     }
 
@@ -151,8 +152,17 @@ int getManifestCodecVersion() {
         }
     }
 
-    private ChecksumBlobStoreFormat<ClusterMetadataManifest> getClusterMetadataManifestBlobStoreFormat() {
+    private ChecksumBlobStoreFormat<ClusterMetadataManifest> getClusterMetadataManifestBlobStoreFormatForDownload() {
         long codecVersion = getManifestCodecVersion();
+        return getClusterMetadataManifestBlobStoreFormat(codecVersion);
+    }
+
+    private ChecksumBlobStoreFormat<ClusterMetadataManifest> getClusterMetadataManifestBlobStoreFormatForUpload() {
+        long codecVersion = clusterMetadataManifest.getCodecVersion();
+        return getClusterMetadataManifestBlobStoreFormat(codecVersion);
+    }
+
+    private ChecksumBlobStoreFormat<ClusterMetadataManifest> getClusterMetadataManifestBlobStoreFormat(long codecVersion) {
         if (codecVersion == ClusterMetadataManifest.MANIFEST_CURRENT_CODEC_VERSION) {
             return CLUSTER_METADATA_MANIFEST_FORMAT;
         } else if (codecVersion == ClusterMetadataManifest.CODEC_V3) {