Handled recovery to start from lowest seqno if pruning is enabled

opensearch-project · Sep 6, 2021 · 6c546a3 · 6c546a3
1 parent acd09db
commit 6c546a3
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 9 deletions.
diff --git a/server/src/main/java/org/opensearch/indices/recovery/RecoverySourceHandler.java b/server/src/main/java/org/opensearch/indices/recovery/RecoverySourceHandler.java
@@ -261,11 +261,24 @@ && isTargetSameHistory()
                 // advances and not when creating a new safe commit. In any case this is a best-effort thing since future recoveries can
                 // always fall back to file-based ones, and only really presents a problem if this primary fails before things have settled
                 // down.
-                startingSeqNo = softDeletesEnabled
-                    ? Long.parseLong(safeCommitRef.getIndexCommit().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)) + 1L
-                    : 0;
+                if(softDeletesEnabled) {
+                    final long minimumRetainingSequenceNumber = shard.getRetentionLeases()
+                        .leases()
+                        .stream()
+                        .mapToLong(RetentionLease::retainingSequenceNumber)
+                        .min()
+                        .orElse(Long.MAX_VALUE);
+                    final long safeCommitSeqNo = Long.parseLong(safeCommitRef.getIndexCommit()
+                        .getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)) + 1L;
+                    if(shard.indexSettings().shouldPruneTranslogByRetentionLease()) {
+                        startingSeqNo = SequenceNumbers.min(safeCommitSeqNo, minimumRetainingSequenceNumber);
+                    } else {
+                        startingSeqNo = safeCommitSeqNo;
+                    }
+                } else {
+                    startingSeqNo = 0;
+                }
                 logger.trace("performing file-based recovery followed by history replay starting at [{}]", startingSeqNo);
-
                 try {
                     final int estimateNumOps = shard.estimateNumberOfHistoryOperations("peer-recovery", historySource, startingSeqNo);
                     final Releasable releaseStore = acquireStore(shard.store());

diff --git a/server/src/test/java/org/opensearch/index/translog/TranslogDeletionPolicyTests.java b/server/src/test/java/org/opensearch/index/translog/TranslogDeletionPolicyTests.java
@@ -90,11 +90,6 @@ public void testBytesRetention() throws IOException {
         }
     }
 
-    public void testBytesRetentionWithRetentionLease() throws Exception {
-        long now = System.currentTimeMillis();
-
-    }
-
     public void testAgeRetention() throws IOException {
         long now = System.currentTimeMillis();
         Tuple<List<TranslogReader>, TranslogWriter> readersAndWriter = createReadersAndWriter(now);