Keeps index commits up to the current global checkpoint

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -25,7 +25,6 @@
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.IndexWriterConfig;
-import org.apache.lucene.index.KeepOnlyLastCommitDeletionPolicy;
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LiveIndexWriterConfig;
 import org.apache.lucene.index.MergePolicy;
@@ -163,12 +162,6 @@ public InternalEngine(EngineConfig engineConfig) {
             maxUnsafeAutoIdTimestamp.set(Long.MAX_VALUE);
         }
         this.uidField = engineConfig.getIndexSettings().isSingleType() ? IdFieldMapper.NAME : UidFieldMapper.NAME;
-        final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
-                engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
-                engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis()
-        );
-        this.deletionPolicy = new CombinedDeletionPolicy(
-                new SnapshotDeletionPolicy(new KeepOnlyLastCommitDeletionPolicy()), translogDeletionPolicy, openMode);
         store.incRef();
         IndexWriter writer = null;
         Translog translog = null;
@@ -183,28 +176,37 @@ public InternalEngine(EngineConfig engineConfig) {
             throttle = new IndexThrottle();
             try {
                 final SeqNoStats seqNoStats;
+                final boolean shouldCreateIndex;
                 switch (openMode) {
                     case OPEN_INDEX_AND_TRANSLOG:
-                        writer = createWriter(false);
                         final long globalCheckpoint = Translog.readGlobalCheckpoint(engineConfig.getTranslogConfig().getTranslogPath());
                         seqNoStats = store.loadSeqNoStats(globalCheckpoint);
+                        shouldCreateIndex = false;
                         break;
                     case OPEN_INDEX_CREATE_TRANSLOG:
-                        writer = createWriter(false);
                         seqNoStats = store.loadSeqNoStats(SequenceNumbers.UNASSIGNED_SEQ_NO);
+                        shouldCreateIndex = false;
                         break;
                     case CREATE_INDEX_AND_TRANSLOG:
-                        writer = createWriter(true);
                         seqNoStats = new SeqNoStats(
                                 SequenceNumbers.NO_OPS_PERFORMED,
                                 SequenceNumbers.NO_OPS_PERFORMED,
                                 SequenceNumbers.UNASSIGNED_SEQ_NO);
+                        shouldCreateIndex = true;
                         break;
                     default:
                         throw new IllegalArgumentException(openMode.toString());
                 }
                 logger.trace("recovered [{}]", seqNoStats);
                 seqNoService = seqNoServiceSupplier.apply(engineConfig, seqNoStats);
+                final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
+                    engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
+                    engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis()
+                );
+                this.deletionPolicy = new CombinedDeletionPolicy(
+                    new SnapshotDeletionPolicy(new KeepUntilGlobalCheckpointDeletionPolicy(seqNoService::getGlobalCheckpoint)),
+                    translogDeletionPolicy, openMode);
+                writer = createWriter(shouldCreateIndex);
                 updateMaxUnsafeAutoIdTimestampFromWriter(writer);
                 historyUUID = loadOrGenerateHistoryUUID(writer, engineConfig.getForceNewHistoryUUID());
                 Objects.requireNonNull(historyUUID, "history uuid should not be null");

core/src/main/java/org/elasticsearch/index/engine/KeepUntilGlobalCheckpointDeletionPolicy.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.engine;
+
+import org.apache.lucene.index.IndexCommit;
+import org.apache.lucene.index.IndexDeletionPolicy;
+import org.elasticsearch.index.seqno.SequenceNumbers;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.LongSupplier;
+
+/**
+ * An {@link IndexDeletionPolicy} keeps the oldest commit whose max sequence number is not
+ * greater than the current global checkpoint, and also keeps all subsequent commits. Once those
+ * commits are kept, a {@link CombinedDeletionPolicy} will retain translog operations at least up to
+ * the current global checkpoint.
+ */
+public final class KeepUntilGlobalCheckpointDeletionPolicy extends IndexDeletionPolicy {
+    private final LongSupplier globalCheckpointSupplier;
+
+    public KeepUntilGlobalCheckpointDeletionPolicy(LongSupplier globalCheckpointSupplier) {
+        this.globalCheckpointSupplier = globalCheckpointSupplier;
+    }
+
+    @Override
+    public void onInit(List<? extends IndexCommit> commits) throws IOException {
+        if (commits.isEmpty() == false) {
+            onCommit(commits);
+        }
+    }
+
+    @Override
+    public void onCommit(List<? extends IndexCommit> commits) throws IOException {
+        assert commits.isEmpty() == false;
+        final int keptIndex = indexOfKeptCommits(commits);
+        for (int i = 0; i < keptIndex; i++) {
+            commits.get(i).delete();
+        }
+        assert commits.get(commits.size() - 1).isDeleted() == false : "The last commit must not be deleted";
+    }
+
+    private int indexOfKeptCommits(List<? extends IndexCommit> commits) throws IOException {
+        final long currentGlobalCheckpoint = globalCheckpointSupplier.getAsLong();
+        // Commits are sorted by age (the 0th one is the oldest commit).
+        for (int i = commits.size() - 1; i >= 0; i--) {
+            final Map<String, String> commitUserData = commits.get(i).getUserData();
+            // Index from 5.x does not contain MAX_SEQ_NO, we should keep either the more recent commit with MAX_SEQ_NO,
+            // or the last commit.
+            if (commitUserData.containsKey(SequenceNumbers.MAX_SEQ_NO) == false) {
+                return Math.min(i + 1, commits.size() - 1);
+            }
+            final long maxSeqNoFromCommit = Long.parseLong(commitUserData.get(SequenceNumbers.MAX_SEQ_NO));
+            if (maxSeqNoFromCommit <= currentGlobalCheckpoint) {
+                return i;
+            }
+        }
+        return -1;
+    }
+
+}