Trim unreferenced translog the safe commit advanced

Since elastic#28140 when the global checkpoint is advanced, we try to move the safe commit forward, and clean old index commits if possible. However, we forget to trim unreferenced translog. This change makes sure that we prune both translog and index commits when the safe commit advanced. Relates elastic#28140 Closes elastic#32089
dnhatn · Aug 17, 2018 · a8594ac · a8594ac
1 parent 647705e
commit a8594ac
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 4 deletions.
diff --git a/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java b/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
@@ -478,6 +478,7 @@ public Translog.Location getTranslogLastWriteLocation() {
     private void revisitIndexDeletionPolicyOnTranslogSynced() throws IOException {
         if (combinedDeletionPolicy.hasUnreferencedCommits()) {
             indexWriter.deleteUnusedFiles();
+            translog.trimUnreferencedReaders();
         }
     }
 

diff --git a/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java
@@ -152,6 +152,7 @@
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogConfig;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
+import org.elasticsearch.test.IndexSettingsModule;
 import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
 
@@ -4342,13 +4343,18 @@ public void testAcquireIndexCommit() throws Exception {
 
     public void testCleanUpCommitsWhenGlobalCheckpointAdvanced() throws Exception {
         IOUtils.close(engine, store);
-        store = createStore();
+        final IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("test",
+            Settings.builder().put(defaultSettings.getSettings())
+                .put(IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.getKey(), -1)
+                .put(IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getKey(), -1).build());
         final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
-        try (InternalEngine engine = createEngine(store, createTempDir(), globalCheckpoint::get)) {
+        try (Store store = createStore();
+             InternalEngine engine =
+                 createEngine(config(indexSettings, store, createTempDir(), newMergePolicy(), null, null, globalCheckpoint::get))) {
             final int numDocs = scaledRandomIntBetween(10, 100);
             for (int docId = 0; docId < numDocs; docId++) {
                 index(engine, docId);
-                if (frequently()) {
+                if (rarely()) {
                     engine.flush(randomBoolean(), randomBoolean());
                 }
             }
@@ -4362,6 +4368,7 @@ public void testCleanUpCommitsWhenGlobalCheckpointAdvanced() throws Exception {
             globalCheckpoint.set(randomLongBetween(engine.getLocalCheckpoint(), Long.MAX_VALUE));
             engine.syncTranslog();
             assertThat(DirectoryReader.listCommits(store.directory()), contains(commits.get(commits.size() - 1)));
+            assertThat(engine.estimateTranslogOperationsFromMinSeq(0L), equalTo(0));
         }
     }
 

diff --git a/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryTests.java b/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryTests.java
@@ -73,7 +73,6 @@ public void testTranslogHistoryTransferred() throws Exception {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/32089")
     public void testRetentionPolicyChangeDuringRecovery() throws Exception {
         try (ReplicationGroup shards = createGroup(0)) {
             shards.startPrimary();