Address PR comments and fix flaky tests

Signed-off-by: Sachin Kale <[email protected]>
opensearch-project · Sep 25, 2024 · af7fcbf · af7fcbf
1 parent 8d3e255
commit af7fcbf
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 5 deletions.
diff --git a/...rTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsGarbageCollectionIT.java b/...rTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsGarbageCollectionIT.java
@@ -167,7 +167,7 @@ public void testLiveIndexNoPinnedTimestampsWithExtraGenSettingWithinLimit() thro
 
         assertBusy(() -> {
             List<Path> metadataFiles = Files.list(translogMetadataPath).collect(Collectors.toList());
-            assertEquals(numDocs + 1, metadataFiles.size());
+            assertTrue(metadataFiles.size() >= numDocs + 1);
 
             verifyTranslogDataFileCount(metadataFiles, translogDataPath);
         }, 30, TimeUnit.SECONDS);

diff --git a/server/src/main/java/org/opensearch/index/translog/RemoteFsTimestampAwareTranslog.java b/server/src/main/java/org/opensearch/index/translog/RemoteFsTimestampAwareTranslog.java
@@ -34,6 +34,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
@@ -61,7 +62,7 @@ public class RemoteFsTimestampAwareTranslog extends RemoteFsTranslog {
     private final Map<String, Tuple<Long, Long>> oldFormatMetadataFileGenerationMap;
     private final Map<String, Tuple<Long, Long>> oldFormatMetadataFilePrimaryTermMap;
     private final AtomicLong minPrimaryTermInRemote = new AtomicLong(Long.MAX_VALUE);
-    private long previousMinRemoteGenReferenced = -1;
+    private final AtomicBoolean triggerTrimOnMinRemoteGenReferencedChange = new AtomicBoolean(false);
 
     public RemoteFsTimestampAwareTranslog(
         TranslogConfig config,
@@ -106,6 +107,11 @@ protected void onDelete() {
         }
     }
 
+    @Override
+    protected void onMinRemoteGenReferencedChange() {
+        triggerTrimOnMinRemoteGenReferencedChange.set(true);
+    }
+
     @Override
     public void trimUnreferencedReaders() throws IOException {
         trimUnreferencedReaders(false, true);
@@ -148,10 +154,10 @@ protected void trimUnreferencedReaders(boolean indexDeleted, boolean trimLocal)
 
         // This code block ensures parity with RemoteFsTranslog. Without this, we will end up making list translog metadata
         // call in each invocation of trimUnreferencedReaders
-        if (indexDeleted == false && previousMinRemoteGenReferenced == minRemoteGenReferenced) {
+        if (indexDeleted == false && triggerTrimOnMinRemoteGenReferencedChange.get() == false) {
             return;
-        } else if (previousMinRemoteGenReferenced != minRemoteGenReferenced) {
-            previousMinRemoteGenReferenced = minRemoteGenReferenced;
+        } else if (triggerTrimOnMinRemoteGenReferencedChange.get()) {
+            triggerTrimOnMinRemoteGenReferencedChange.set(false);
         }
 
         // Since remote generation deletion is async, this ensures that only one generation deletion happens at a time.

diff --git a/server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java b/server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java
@@ -683,7 +683,11 @@ private class RemoteFsTranslogTransferListener implements TranslogTransferListen
         @Override
         public void onUploadComplete(TransferSnapshot transferSnapshot) throws IOException {
             maxRemoteTranslogGenerationUploaded = generation;
+            long previousMinRemoteGenReferenced = minRemoteGenReferenced;
             minRemoteGenReferenced = getMinFileGeneration();
+            if (previousMinRemoteGenReferenced != minRemoteGenReferenced) {
+                onMinRemoteGenReferencedChange();
+            }
             logger.debug(
                 "Successfully uploaded translog for primary term = {}, generation = {}, maxSeqNo = {}, minRemoteGenReferenced = {}",
                 primaryTerm,
@@ -703,6 +707,10 @@ public void onUploadFailed(TransferSnapshot transferSnapshot, Exception ex) thro
         }
     }
 
+    protected void onMinRemoteGenReferencedChange() {
+
+    }
+
     @Override
     public long getMinUnreferencedSeqNoInSegments(long minUnrefCheckpointInLastCommit) {
         return minSeqNoToKeep;