Move the optimisation to upload Metadata method

Signed-off-by: sjs004 <simarjeet.singh004@gmail.com>

Author: sjs004

server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java

@@ -91,6 +91,9 @@ public final class RemoteStoreRefreshListener extends ReleasableRetryableRefresh
     private final SegmentReplicationCheckpointPublisher checkpointPublisher;
     private final RemoteStoreSettings remoteStoreSettings;
     private final RemoteStoreUploader remoteStoreUploader;
+    private volatile long lastUploadedPrimaryTerm = INVALID_PRIMARY_TERM; // Use constant or -1
+    private volatile long lastUploadedGeneration = -1;
+    private volatile long lastUploadedTranslogGeneration = -1;
 
     public RemoteStoreRefreshListener(
         IndexShard indexShard,
@@ -430,27 +433,40 @@ private boolean isRefreshAfterCommitSafe() {
 
     void uploadMetadata(Collection<String> localSegmentsPostRefresh, SegmentInfos segmentInfos, ReplicationCheckpoint replicationCheckpoint)
         throws IOException {
+
         final long maxSeqNo = ((InternalEngine) indexShard.getEngine()).currentOngoingRefreshCheckpoint();
+        Translog.TranslogGeneration translogGeneration = indexShard.getEngine().translogManager().getTranslogGeneration();
+        if (translogGeneration == null) {
+            throw new UnsupportedOperationException("Encountered null TranslogGeneration while uploading metadata to remote segment store");
+        }
+        long translogFileGeneration = translogGeneration.translogFileGeneration;
+        if (this.lastUploadedPrimaryTerm == replicationCheckpoint.getPrimaryTerm()
+            && this.lastUploadedGeneration == segmentInfos.getGeneration()
+            && this.lastUploadedTranslogGeneration == translogFileGeneration) {
+
+            logger.debug("Skipping metadata upload (deduplicated) - state is already persisted.");
+            return;
+        }
+
         SegmentInfos segmentInfosSnapshot = segmentInfos.clone();
         Map<String, String> userData = segmentInfosSnapshot.getUserData();
         userData.put(LOCAL_CHECKPOINT_KEY, String.valueOf(maxSeqNo));
         userData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo));
         segmentInfosSnapshot.setUserData(userData, false);
 
-        Translog.TranslogGeneration translogGeneration = indexShard.getEngine().translogManager().getTranslogGeneration();
-        if (translogGeneration == null) {
-            throw new UnsupportedOperationException("Encountered null TranslogGeneration while uploading metadata to remote segment store");
-        } else {
-            long translogFileGeneration = translogGeneration.translogFileGeneration;
-            remoteDirectory.uploadMetadata(
-                localSegmentsPostRefresh,
-                segmentInfosSnapshot,
-                storeDirectory,
-                translogFileGeneration,
-                replicationCheckpoint,
-                indexShard.getNodeId()
-            );
-        }
+        remoteDirectory.uploadMetadata(
+            localSegmentsPostRefresh,
+            segmentInfosSnapshot,
+            storeDirectory,
+            translogFileGeneration,
+            replicationCheckpoint,
+            indexShard.getNodeId()
+        );
+
+        // Update the Listener's cache on success
+        this.lastUploadedPrimaryTerm = replicationCheckpoint.getPrimaryTerm();
+        this.lastUploadedGeneration = segmentInfos.getGeneration();
+        this.lastUploadedTranslogGeneration = translogFileGeneration;
     }
 
     boolean isLowPriorityUpload() {