Add logging to index commit deletion policy (#28448)

This would help us to figure out which index commit that an engine
started with or used in peer-recovery.

Relates #28405

Author: dnhatn

server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.index.engine;
 
 import com.carrotsearch.hppc.ObjectIntHashMap;
+import org.apache.logging.log4j.Logger;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.IndexDeletionPolicy;
 import org.apache.lucene.store.Directory;
@@ -31,6 +32,7 @@
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.function.LongSupplier;
 
@@ -42,6 +44,7 @@
  * the current global checkpoint except the index commit which has the highest max sequence number among those.
  */
 public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
+    private final Logger logger;
     private final TranslogDeletionPolicy translogDeletionPolicy;
     private final EngineConfig.OpenMode openMode;
     private final LongSupplier globalCheckpointSupplier;
@@ -50,9 +53,10 @@ public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
     private volatile IndexCommit safeCommit; // the most recent safe commit point - its max_seqno at most the persisted global checkpoint.
     private volatile IndexCommit lastCommit; // the most recent commit point
 
-    CombinedDeletionPolicy(EngineConfig.OpenMode openMode, TranslogDeletionPolicy translogDeletionPolicy,
+    CombinedDeletionPolicy(EngineConfig.OpenMode openMode, Logger logger, TranslogDeletionPolicy translogDeletionPolicy,
                            LongSupplier globalCheckpointSupplier, IndexCommit startingCommit) {
         this.openMode = openMode;
+        this.logger = logger;
         this.translogDeletionPolicy = translogDeletionPolicy;
         this.globalCheckpointSupplier = globalCheckpointSupplier;
         this.startingCommit = startingCommit;
@@ -104,8 +108,12 @@ public synchronized void onInit(List<? extends IndexCommit> commits) throws IOEx
      * (v6.2), may not have a safe commit. If that index has a snapshotted commit without translog and an unsafe commit,
      * the policy can consider the snapshotted commit as a safe commit for recovery even the commit does not have translog.
      */
-    private void keepOnlyStartingCommitOnInit(List<? extends IndexCommit> commits) {
-        commits.stream().filter(commit -> startingCommit.equals(commit) == false).forEach(IndexCommit::delete);
+    private void keepOnlyStartingCommitOnInit(List<? extends IndexCommit> commits) throws IOException {
+        for (IndexCommit commit : commits) {
+            if (startingCommit.equals(commit) == false) {
+                this.deleteCommit(commit);
+            }
+        }
         assert startingCommit.isDeleted() == false : "Starting commit must not be deleted";
         lastCommit = startingCommit;
         safeCommit = startingCommit;
@@ -118,14 +126,22 @@ public synchronized void onCommit(List<? extends IndexCommit> commits) throws IO
         safeCommit = commits.get(keptPosition);
         for (int i = 0; i < keptPosition; i++) {
             if (snapshottedCommits.containsKey(commits.get(i)) == false) {
-                commits.get(i).delete();
+                deleteCommit(commits.get(i));
             }
         }
         updateTranslogDeletionPolicy();
     }
 
+    private void deleteCommit(IndexCommit commit) throws IOException {
+        assert commit.isDeleted() == false : "Index commit [" + commitDescription(commit) + "] is deleted twice";
+        logger.debug("Delete index commit [{}]", commitDescription(commit));
+        commit.delete();
+        assert commit.isDeleted() : "Deletion commit [" + commitDescription(commit) + "] was suppressed";
+    }
+
     private void updateTranslogDeletionPolicy() throws IOException {
         assert Thread.holdsLock(this);
+        logger.debug("Safe commit [{}], last commit [{}]", commitDescription(safeCommit), commitDescription(lastCommit));
         assert safeCommit.isDeleted() == false : "The safe commit must not be deleted";
         final long minRequiredGen = Long.parseLong(safeCommit.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
         assert lastCommit.isDeleted() == false : "The last commit must not be deleted";
@@ -239,6 +255,13 @@ boolean hasUnreferencedCommits() throws IOException {
         return false;
     }
 
+    /**
+     * Returns a description for a given {@link IndexCommit}. This should be only used for logging and debugging.
+     */
+    public static String commitDescription(IndexCommit commit) throws IOException {
+        return String.format(Locale.ROOT, "CommitPoint{segment[%s], userData[%s]}", commit.getSegmentsFileName(), commit.getUserData());
+    }
+
     /**
      * A wrapper of an index commit that prevents it from being deleted.
      */

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

@@ -185,7 +185,7 @@ public InternalEngine(EngineConfig engineConfig) {
                 translog = openTranslog(engineConfig, translogDeletionPolicy, engineConfig.getGlobalCheckpointSupplier(), startingCommit);
                 assert translog.getGeneration() != null;
                 this.translog = translog;
-                this.combinedDeletionPolicy = new CombinedDeletionPolicy(openMode, translogDeletionPolicy,
+                this.combinedDeletionPolicy = new CombinedDeletionPolicy(openMode, logger, translogDeletionPolicy,
                     translog::getLastSyncedGlobalCheckpoint, startingCommit);
                 writer = createWriter(openMode == EngineConfig.OpenMode.CREATE_INDEX_AND_TRANSLOG, startingCommit);
                 updateMaxUnsafeAutoIdTimestampFromWriter(writer);

server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.indices.recovery;
 
+import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
 import org.apache.lucene.index.DirectoryReader;
@@ -64,6 +65,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.StringJoiner;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -320,7 +322,7 @@ private StartRecoveryRequest getStartRecoveryRequest(final RecoveryTarget recove
 
         final long startingSeqNo;
         if (metadataSnapshot.size() > 0) {
-            startingSeqNo = getStartingSeqNo(recoveryTarget);
+            startingSeqNo = getStartingSeqNo(logger, recoveryTarget);
         } else {
             startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
         }
@@ -354,12 +356,20 @@ private StartRecoveryRequest getStartRecoveryRequest(final RecoveryTarget recove
      * @return the starting sequence number or {@link SequenceNumbers#UNASSIGNED_SEQ_NO} if obtaining the starting sequence number
      * failed
      */
-    public static long getStartingSeqNo(final RecoveryTarget recoveryTarget) {
+    public static long getStartingSeqNo(final Logger logger, final RecoveryTarget recoveryTarget) {
         try {
             final long globalCheckpoint = Translog.readGlobalCheckpoint(recoveryTarget.translogLocation());
             final List<IndexCommit> existingCommits = DirectoryReader.listCommits(recoveryTarget.store().directory());
             final IndexCommit safeCommit = CombinedDeletionPolicy.findSafeCommitPoint(existingCommits, globalCheckpoint);
             final SequenceNumbers.CommitInfo seqNoStats = recoveryTarget.store().loadSeqNoInfo(safeCommit);
+            if (logger.isTraceEnabled()) {
+                final StringJoiner descriptionOfExistingCommits = new StringJoiner(",");
+                for (IndexCommit commit : existingCommits) {
+                    descriptionOfExistingCommits.add(CombinedDeletionPolicy.commitDescription(commit));
+                }
+                logger.trace("Calculate starting seqno based on global checkpoint [{}], safe commit [{}], existing commits [{}]",
+                    globalCheckpoint, CombinedDeletionPolicy.commitDescription(safeCommit), descriptionOfExistingCommits);
+            }
             if (seqNoStats.maxSeqNo <= globalCheckpoint) {
                 assert seqNoStats.localCheckpoint <= globalCheckpoint;
                 /*

server/src/test/java/org/elasticsearch/index/engine/CombinedDeletionPolicyTests.java

@@ -56,7 +56,7 @@ public void testKeepCommitsAfterGlobalCheckpoint() throws Exception {
         final AtomicLong globalCheckpoint = new AtomicLong();
         TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
 
         final LongArrayList maxSeqNoList = new LongArrayList();
         final LongArrayList translogGenList = new LongArrayList();
@@ -96,7 +96,7 @@ public void testAcquireIndexCommit() throws Exception {
         final UUID translogUUID = UUID.randomUUID();
         TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
         long lastMaxSeqNo = between(1, 1000);
         long lastTranslogGen = between(1, 20);
         int safeIndex = 0;
@@ -117,6 +117,7 @@ public void testAcquireIndexCommit() throws Exception {
             long upper = safeIndex == commitList.size() - 1 ? lastMaxSeqNo :
                 Long.parseLong(commitList.get(safeIndex + 1).getUserData().get(SequenceNumbers.MAX_SEQ_NO)) - 1;
             globalCheckpoint.set(randomLongBetween(lower, upper));
+            commitList.forEach(this::resetDeletion);
             indexPolicy.onCommit(commitList);
             // Captures and releases some commits
             int captures = between(0, 5);
@@ -145,6 +146,7 @@ public void testAcquireIndexCommit() throws Exception {
         }
         snapshottingCommits.forEach(indexPolicy::releaseCommit);
         globalCheckpoint.set(randomLongBetween(lastMaxSeqNo, Long.MAX_VALUE));
+        commitList.forEach(this::resetDeletion);
         indexPolicy.onCommit(commitList);
         for (int i = 0; i < commitList.size() - 1; i++) {
             assertThat(commitList.get(i).isDeleted(), equalTo(true));
@@ -160,7 +162,7 @@ public void testLegacyIndex() throws Exception {
 
         TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
 
         long legacyTranslogGen = randomNonNegativeLong();
         IndexCommit legacyCommit = mockLegacyIndexCommit(translogUUID, legacyTranslogGen);
@@ -181,6 +183,7 @@ public void testLegacyIndex() throws Exception {
         assertThat(translogPolicy.getTranslogGenerationOfLastCommit(), equalTo(safeTranslogGen));
 
         // Make the fresh commit safe.
+        resetDeletion(legacyCommit);
         globalCheckpoint.set(randomLongBetween(maxSeqNo, Long.MAX_VALUE));
         indexPolicy.onCommit(Arrays.asList(legacyCommit, freshCommit));
         verify(legacyCommit, times(2)).delete();
@@ -194,7 +197,7 @@ public void testKeepSingleNoOpsCommits() throws Exception {
         final UUID translogUUID = UUID.randomUUID();
         TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
 
         final List<IndexCommit> commitList = new ArrayList<>();
         final int numOfNoOpsCommits = between(1, 10);
@@ -222,6 +225,7 @@ public void testKeepSingleNoOpsCommits() throws Exception {
         }
         // If the global checkpoint is still unassigned, we should still keep one NO_OPS_PERFORMED commit.
         globalCheckpoint.set(SequenceNumbers.UNASSIGNED_SEQ_NO);
+        commitList.forEach(this::resetDeletion);
         indexPolicy.onCommit(commitList);
         assertThat(translogPolicy.getMinTranslogGenerationForRecovery(), equalTo(lastNoopTranslogGen));
         assertThat(translogPolicy.getTranslogGenerationOfLastCommit(), equalTo(lastTranslogGen));
@@ -232,6 +236,7 @@ public void testKeepSingleNoOpsCommits() throws Exception {
         // Delete no-ops commit if global checkpoint advanced enough.
         final long lower = Long.parseLong(commitList.get(numOfNoOpsCommits).getUserData().get(SequenceNumbers.MAX_SEQ_NO));
         globalCheckpoint.set(randomLongBetween(lower, Long.MAX_VALUE));
+        commitList.forEach(this::resetDeletion);
         indexPolicy.onCommit(commitList);
         assertThat(translogPolicy.getMinTranslogGenerationForRecovery(), greaterThan(lastNoopTranslogGen));
         assertThat(translogPolicy.getTranslogGenerationOfLastCommit(), equalTo(lastTranslogGen));
@@ -242,7 +247,7 @@ public void testDeleteInvalidCommits() throws Exception {
         final AtomicLong globalCheckpoint = new AtomicLong(randomNonNegativeLong());
         TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_CREATE_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_CREATE_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
 
         final int invalidCommits = between(1, 10);
         final List<IndexCommit> commitList = new ArrayList<>();
@@ -280,7 +285,7 @@ public void testKeepOnlyStartingCommitOnInit() throws Exception {
         }
         final IndexCommit startingCommit = randomFrom(commitList);
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, startingCommit);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, startingCommit);
         indexPolicy.onInit(commitList);
         for (IndexCommit commit : commitList) {
             if (commit.equals(startingCommit) == false) {
@@ -299,7 +304,7 @@ public void testCheckUnreferencedCommits() throws Exception {
         final UUID translogUUID = UUID.randomUUID();
         final TranslogDeletionPolicy translogPolicy = createTranslogDeletionPolicy();
         CombinedDeletionPolicy indexPolicy = new CombinedDeletionPolicy(
-            OPEN_INDEX_AND_TRANSLOG, translogPolicy, globalCheckpoint::get, null);
+            OPEN_INDEX_AND_TRANSLOG, logger, translogPolicy, globalCheckpoint::get, null);
         final List<IndexCommit> commitList = new ArrayList<>();
         int totalCommits = between(2, 20);
         long lastMaxSeqNo = between(1, 1000);
@@ -311,6 +316,7 @@ public void testCheckUnreferencedCommits() throws Exception {
         }
         IndexCommit safeCommit = randomFrom(commitList);
         globalCheckpoint.set(Long.parseLong(safeCommit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)));
+        commitList.forEach(this::resetDeletion);
         indexPolicy.onCommit(commitList);
         if (safeCommit == commitList.get(commitList.size() - 1)) {
             // Safe commit is the last commit - no need to clean up
@@ -324,6 +330,7 @@ public void testCheckUnreferencedCommits() throws Exception {
             // Advanced enough
             globalCheckpoint.set(randomLongBetween(lastMaxSeqNo, Long.MAX_VALUE));
             assertThat(indexPolicy.hasUnreferencedCommits(), equalTo(true));
+            commitList.forEach(this::resetDeletion);
             indexPolicy.onCommit(commitList);
             // Safe commit is the last commit - no need to clean up
             assertThat(translogPolicy.getMinTranslogGenerationForRecovery(), equalTo(lastTranslogGen));
@@ -337,17 +344,21 @@ IndexCommit mockIndexCommit(long maxSeqNo, UUID translogUUID, long translogGen)
         userData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo));
         userData.put(Translog.TRANSLOG_UUID_KEY, translogUUID.toString());
         userData.put(Translog.TRANSLOG_GENERATION_KEY, Long.toString(translogGen));
-        final AtomicBoolean deleted = new AtomicBoolean();
         final IndexCommit commit = mock(IndexCommit.class);
         final Directory directory = mock(Directory.class);
         when(commit.getUserData()).thenReturn(userData);
         when(commit.getDirectory()).thenReturn(directory);
+        resetDeletion(commit);
+        return commit;
+    }
+
+    void resetDeletion(IndexCommit commit) {
+        final AtomicBoolean deleted = new AtomicBoolean();
         when(commit.isDeleted()).thenAnswer(args -> deleted.get());
         doAnswer(arg -> {
             deleted.set(true);
             return null;
         }).when(commit).delete();
-        return commit;
     }
 
     IndexCommit mockLegacyIndexCommit(UUID translogUUID, long translogGen) throws IOException {
@@ -356,6 +367,7 @@ IndexCommit mockLegacyIndexCommit(UUID translogUUID, long translogGen) throws IO
         userData.put(Translog.TRANSLOG_GENERATION_KEY, Long.toString(translogGen));
         final IndexCommit commit = mock(IndexCommit.class);
         when(commit.getUserData()).thenReturn(userData);
+        resetDeletion(commit);
         return commit;
     }
 }

server/src/test/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetServiceTests.java

@@ -33,7 +33,7 @@ public void testGetStartingSeqNo() throws Exception {
             {
                 recoveryEmptyReplica(replica);
                 final RecoveryTarget recoveryTarget = new RecoveryTarget(replica, null, null, null);
-                assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(0L));
+                assertThat(PeerRecoveryTargetService.getStartingSeqNo(logger, recoveryTarget), equalTo(0L));
                 recoveryTarget.decRef();
             }
             // Last commit is good - use it.
@@ -49,7 +49,7 @@ public void testGetStartingSeqNo() throws Exception {
                 replica.updateGlobalCheckpointOnReplica(initDocs - 1, "test");
                 replica.getTranslog().sync();
                 final RecoveryTarget recoveryTarget = new RecoveryTarget(replica, null, null, null);
-                assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(initDocs));
+                assertThat(PeerRecoveryTargetService.getStartingSeqNo(logger, recoveryTarget), equalTo(initDocs));
                 recoveryTarget.decRef();
             }
             // Global checkpoint does not advance, last commit is not good - use the previous commit
@@ -63,15 +63,15 @@ public void testGetStartingSeqNo() throws Exception {
                 }
                 flushShard(replica);
                 final RecoveryTarget recoveryTarget = new RecoveryTarget(replica, null, null, null);
-                assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(initDocs));
+                assertThat(PeerRecoveryTargetService.getStartingSeqNo(logger, recoveryTarget), equalTo(initDocs));
                 recoveryTarget.decRef();
             }
             // Advances the global checkpoint, a safe commit also advances
             {
                 replica.updateGlobalCheckpointOnReplica(initDocs + moreDocs - 1, "test");
                 replica.getTranslog().sync();
                 final RecoveryTarget recoveryTarget = new RecoveryTarget(replica, null, null, null);
-                assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(initDocs + moreDocs));
+                assertThat(PeerRecoveryTargetService.getStartingSeqNo(logger, recoveryTarget), equalTo(initDocs + moreDocs));
                 recoveryTarget.decRef();
             }
         } finally {

test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java

@@ -463,7 +463,7 @@ protected final void recoverReplica(final IndexShard replica,
         final Store.MetadataSnapshot snapshot = getMetadataSnapshotOrEmpty(replica);
         final long startingSeqNo;
         if (snapshot.size() > 0) {
-            startingSeqNo = PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget);
+            startingSeqNo = PeerRecoveryTargetService.getStartingSeqNo(logger, recoveryTarget);
         } else {
             startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
         }