Adjust CombinedDeletionPolicy for multiple commits (#27456)

dnhatn · web-flow · commit e95d18ec23e9 · 2017-11-23T11:34:50.000-05:00
Today, we keep only the last index commit and use only it to calculate the minimum required translog generation. This may no longer be correct as we introduced a new deletion policy which keeps multiple index commits. This change adjusts the CombinedDeletionPolicy so that it can work correctly with a new index deletion policy. Relates to #10708, #27367
diff --git a/core/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java b/core/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java
@@ -71,12 +71,18 @@ public void onCommit(List<? extends IndexCommit> commits) throws IOException {
     }
 
     private void setLastCommittedTranslogGeneration(List<? extends IndexCommit> commits) throws IOException {
-        // when opening an existing lucene index, we currently always open the last commit.
-        // we therefore use the translog gen as the one that will be required for recovery
-        final IndexCommit indexCommit = commits.get(commits.size() - 1);
-        assert indexCommit.isDeleted() == false : "last commit is deleted";
-        long minGen = Long.parseLong(indexCommit.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
-        translogDeletionPolicy.setMinTranslogGenerationForRecovery(minGen);
+        // We need to keep translog since the smallest translog generation of un-deleted commits.
+        // However, there are commits that are not deleted just because they are being snapshotted (rather than being kept by the policy).
+        // TODO: We need to distinguish those commits and skip them in calculating the minimum required translog generation.
+        long minRequiredGen = Long.MAX_VALUE;
+        for (IndexCommit indexCommit : commits) {
+            if (indexCommit.isDeleted() == false) {
+                long translogGen = Long.parseLong(indexCommit.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
+                minRequiredGen = Math.min(translogGen, minRequiredGen);
+            }
+        }
+        assert minRequiredGen != Long.MAX_VALUE : "All commits are deleted";
+        translogDeletionPolicy.setMinTranslogGenerationForRecovery(minRequiredGen);
     }
 
     public SnapshotDeletionPolicy getIndexDeletionPolicy() {
diff --git a/core/src/test/java/org/elasticsearch/index/engine/CombinedDeletionPolicyTests.java b/core/src/test/java/org/elasticsearch/index/engine/CombinedDeletionPolicyTests.java
@@ -60,20 +60,23 @@ public void testSettingMinTranslogGen() throws IOException {
             EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG);
         List<IndexCommit> commitList = new ArrayList<>();
         long count = randomIntBetween(10, 20);
-        long lastGen = 0;
+        long minGen = Long.MAX_VALUE;
         for (int i = 0; i < count; i++) {
-            lastGen += randomIntBetween(10, 20000);
+            long lastGen = randomIntBetween(10, 20000);
+            minGen = Math.min(minGen, lastGen);
             commitList.add(mockIndexCommitWithTranslogGen(lastGen));
         }
         combinedDeletionPolicy.onInit(commitList);
-        verify(translogDeletionPolicy, times(1)).setMinTranslogGenerationForRecovery(lastGen);
+        verify(translogDeletionPolicy, times(1)).setMinTranslogGenerationForRecovery(minGen);
         commitList.clear();
+        minGen = Long.MAX_VALUE;
         for (int i = 0; i < count; i++) {
-            lastGen += randomIntBetween(10, 20000);
+            long lastGen = randomIntBetween(10, 20000);
+            minGen = Math.min(minGen, lastGen);
             commitList.add(mockIndexCommitWithTranslogGen(lastGen));
         }
         combinedDeletionPolicy.onCommit(commitList);
-        verify(translogDeletionPolicy, times(1)).setMinTranslogGenerationForRecovery(lastGen);
+        verify(translogDeletionPolicy, times(1)).setMinTranslogGenerationForRecovery(minGen);
     }
 
     IndexCommit mockIndexCommitWithTranslogGen(long gen) throws IOException {
