Abstract out RemoteFsTranslogWithPinnedTimestamps (#15579) (#15623)

* Abstract out RemoteFsTranslog with pinned timestamps


(cherry picked from commit deeb2de)

Signed-off-by: Sachin Kale <[email protected]>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: opensearch-trigger-bot[bot]

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

@@ -69,19 +69,35 @@ public Translog newTranslog(
 
         assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
         BlobStoreRepository blobStoreRepository = ((BlobStoreRepository) repository);
-        return new RemoteFsTranslog(
-            config,
-            translogUUID,
-            deletionPolicy,
-            globalCheckpointSupplier,
-            primaryTermSupplier,
-            persistedSequenceNumberConsumer,
-            blobStoreRepository,
-            threadPool,
-            startedPrimarySupplier,
-            remoteTranslogTransferTracker,
-            remoteStoreSettings
-        );
+        if (RemoteStoreSettings.isPinnedTimestampsEnabled()) {
+            return new RemoteFsTimestampAwareTranslog(
+                config,
+                translogUUID,
+                deletionPolicy,
+                globalCheckpointSupplier,
+                primaryTermSupplier,
+                persistedSequenceNumberConsumer,
+                blobStoreRepository,
+                threadPool,
+                startedPrimarySupplier,
+                remoteTranslogTransferTracker,
+                remoteStoreSettings
+            );
+        } else {
+            return new RemoteFsTranslog(
+                config,
+                translogUUID,
+                deletionPolicy,
+                globalCheckpointSupplier,
+                primaryTermSupplier,
+                persistedSequenceNumberConsumer,
+                blobStoreRepository,
+                threadPool,
+                startedPrimarySupplier,
+                remoteTranslogTransferTracker,
+                remoteStoreSettings
+            );
+        }
     }
 
     public Repository getRepository() {