policy with retaining seqno

dnhatn · dnhatn · commit bda80cde2422 · 2019-12-19T03:00:27.000-05:00
diff --git a/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java b/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
@@ -87,6 +87,7 @@
 import org.elasticsearch.index.merge.MergeStats;
 import org.elasticsearch.index.merge.OnGoingMerge;
 import org.elasticsearch.index.seqno.LocalCheckpointTracker;
+import org.elasticsearch.index.seqno.RetentionLease;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.ElasticsearchMergePolicy;
@@ -189,11 +190,7 @@ public InternalEngine(EngineConfig engineConfig) {
             final EngineConfig engineConfig,
             final BiFunction<Long, Long, LocalCheckpointTracker> localCheckpointTrackerSupplier) {
         super(engineConfig);
-        final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
-                engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
-                engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis(),
-                engineConfig.getIndexSettings().getTranslogRetentionTotalFiles()
-        );
+        final TranslogDeletionPolicy translogDeletionPolicy = newTranslogDeletionPolicy(engineConfig);
         store.incRef();
         IndexWriter writer = null;
         Translog translog = null;
@@ -488,6 +485,22 @@ private void recoverFromTranslogInternal(TranslogRecoveryRunner translogRecovery
         translog.trimUnreferencedReaders();
     }
 
+    private static TranslogDeletionPolicy newTranslogDeletionPolicy(EngineConfig config) {
+        final LongSupplier retainingSeqNo;
+        if (config.getIndexSettings().isSoftDeleteEnabled()) {
+            retainingSeqNo = () -> Long.MAX_VALUE;
+        } else {
+            retainingSeqNo = () -> config.retentionLeasesSupplier().get().leases().stream()
+                .mapToLong(RetentionLease::retainingSequenceNumber)
+                .max().orElse(Long.MAX_VALUE);
+        };
+        return new TranslogDeletionPolicy(
+            config.getIndexSettings().getTranslogRetentionSize().getBytes(),
+            config.getIndexSettings().getTranslogRetentionAge().getMillis(),
+            config.getIndexSettings().getTranslogRetentionTotalFiles(),
+            retainingSeqNo);
+    }
+
     private Translog openTranslog(EngineConfig engineConfig, TranslogDeletionPolicy translogDeletionPolicy,
                                   LongSupplier globalCheckpointSupplier, LongConsumer persistedSequenceNumberConsumer) throws IOException {
 
diff --git a/server/src/main/java/org/elasticsearch/index/translog/TranslogDeletionPolicy.java b/server/src/main/java/org/elasticsearch/index/translog/TranslogDeletionPolicy.java
@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
 
 public class TranslogDeletionPolicy {
 
@@ -65,10 +66,14 @@ public void assertNoOpenTranslogRefs() {
 
     private int retentionTotalFiles;
 
-    public TranslogDeletionPolicy(long retentionSizeInBytes, long retentionAgeInMillis, int retentionTotalFiles) {
+    private final LongSupplier retainingSeqNoSupplier;
+
+    public TranslogDeletionPolicy(long retentionSizeInBytes, long retentionAgeInMillis, int retentionTotalFiles,
+                                  LongSupplier retainingSeqNoSupplier) {
         this.retentionSizeInBytes = retentionSizeInBytes;
         this.retentionAgeInMillis = retentionAgeInMillis;
         this.retentionTotalFiles = retentionTotalFiles;
+        this.retainingSeqNoSupplier = retainingSeqNoSupplier;
         if (Assertions.ENABLED) {
             openTranslogRef = new ConcurrentHashMap<>();
         } else {
@@ -172,7 +177,9 @@ synchronized long minTranslogGenRequired(List<TranslogReader> readers, TranslogW
             minByAgeAndSize = Math.max(minByAge, minBySize);
         }
         long minByNumFiles = getMinTranslogGenByTotalFiles(readers, writer, retentionTotalFiles);
-        return Math.min(Math.max(minByAgeAndSize, minByNumFiles), Math.min(minByLocks, minTranslogGenerationForRecovery));
+        long minByPolicy = Math.max(minByAgeAndSize, minByNumFiles);
+        long minByRetainingSeqNo = getMinTranslogGenByRetainingSeqNo(readers, writer, retainingSeqNoSupplier.getAsLong());
+        return Math.min(Math.max(minByPolicy, minByRetainingSeqNo), Math.min(minByLocks, minTranslogGenerationForRecovery));
     }
 
     static long getMinTranslogGenBySize(List<TranslogReader> readers, TranslogWriter writer, long retentionSizeInBytes) {
@@ -214,6 +221,15 @@ static long getMinTranslogGenByTotalFiles(List<TranslogReader> readers, Translog
         return minGen;
     }
 
+    static long getMinTranslogGenByRetainingSeqNo(List<TranslogReader> readers, TranslogWriter writer, long seqNo) {
+        for (TranslogReader reader : readers) {
+            if (reader.getCheckpoint().maxEffectiveSeqNo() >= seqNo) {
+                return reader.generation;
+            }
+        }
+        return writer.generation;
+    }
+
     protected long currentTime() {
         return System.currentTimeMillis();
     }
