Log more info with max seq no on upload to and download from remote t…

…ranslog (opensearch-project#10973)

* Log more info with max seq no on upload to and download from remote translog

Signed-off-by: Sachin Kale <kalsac@amazon.com>

* Change log level to debug

Signed-off-by: Sachin Kale <kalsac@amazon.com>

* Spotless fixes

Signed-off-by: Sachin Kale <kalsac@amazon.com>

* Fix UTs

Signed-off-by: Sachin Kale <kalsac@amazon.com>

---------

Signed-off-by: Sachin Kale <kalsac@amazon.com>
Co-authored-by: Sachin Kale <kalsac@amazon.com>

server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java

@@ -103,6 +103,7 @@ public RemoteFsTranslog(
         try {
             download(translogTransferManager, location, logger);
             Checkpoint checkpoint = readCheckpoint(location);
+            logger.info("Downloaded data from remote translog till maxSeqNo = {}", checkpoint.maxSeqNo);
             this.readers.addAll(recoverFromFiles(checkpoint));
             if (readers.isEmpty()) {
                 String errorMsg = String.format(Locale.ROOT, "%s at least one reader must be recovered", shardId);
@@ -266,9 +267,13 @@ public void rollGeneration() throws IOException {
     }
 
     private boolean prepareAndUpload(Long primaryTerm, Long generation) throws IOException {
+        long maxSeqNo = -1;
         try (Releasable ignored = writeLock.acquire()) {
             if (generation == null || generation == current.getGeneration()) {
                 try {
+                    if (closed.get() == false) {
+                        maxSeqNo = getMaxSeqNo();
+                    }
                     final TranslogReader reader = current.closeIntoReader();
                     readers.add(reader);
                     copyCheckpointTo(location.resolve(getCommitCheckpointFileName(current.getGeneration())));
@@ -300,17 +305,17 @@ private boolean prepareAndUpload(Long primaryTerm, Long generation) throws IOExc
             // is not updated in remote translog except in primary to primary recovery.
             if (generation == null) {
                 if (closed.get() == false) {
-                    return upload(primaryTerm, current.getGeneration() - 1);
+                    return upload(primaryTerm, current.getGeneration() - 1, maxSeqNo);
                 } else {
-                    return upload(primaryTerm, current.getGeneration());
+                    return upload(primaryTerm, current.getGeneration(), maxSeqNo);
                 }
             } else {
-                return upload(primaryTerm, generation);
+                return upload(primaryTerm, generation, maxSeqNo);
             }
         }
     }
 
-    private boolean upload(Long primaryTerm, Long generation) throws IOException {
+    private boolean upload(long primaryTerm, long generation, long maxSeqNo) throws IOException {
         // During primary relocation (primary-primary peer recovery), both the old and the new primary have engine
         // created with the RemoteFsTranslog. Both primaries are equipped to upload the translogs. The primary mode check
         // below ensures that the real primary only is uploading. Before the primary mode is set as true for the new
@@ -334,7 +339,7 @@ private boolean upload(Long primaryTerm, Long generation) throws IOException {
         ) {
             return translogTransferManager.transferSnapshot(
                 transferSnapshotProvider,
-                new RemoteFsTranslogTransferListener(generation, primaryTerm)
+                new RemoteFsTranslogTransferListener(generation, primaryTerm, maxSeqNo)
             );
         }
 
@@ -522,23 +527,31 @@ private class RemoteFsTranslogTransferListener implements TranslogTransferListen
         /**
          * Generation for the translog
          */
-        private final Long generation;
+        private final long generation;
 
         /**
          * Primary Term for the translog
          */
-        private final Long primaryTerm;
+        private final long primaryTerm;
+
+        private final long maxSeqNo;
 
-        RemoteFsTranslogTransferListener(Long generation, Long primaryTerm) {
+        RemoteFsTranslogTransferListener(long generation, long primaryTerm, long maxSeqNo) {
             this.generation = generation;
             this.primaryTerm = primaryTerm;
+            this.maxSeqNo = maxSeqNo;
         }
 
         @Override
         public void onUploadComplete(TransferSnapshot transferSnapshot) throws IOException {
             maxRemoteTranslogGenerationUploaded = generation;
             minRemoteGenReferenced = getMinFileGeneration();
-            logger.trace("uploaded translog for {} {} ", primaryTerm, generation);
+            logger.debug(
+                "Successfully uploaded translog for primary term = {}, generation = {}, maxSeqNo = {}",
+                primaryTerm,
+                generation,
+                maxSeqNo
+            );
         }
 
         @Override