RATIS-2056. Update purge index when snapshot installed. (apache#1065)

szetszwo · Jun 16, 2024 · b9d0075 · b9d0075
1 parent 67a0c6b
commit b9d0075
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 5 deletions.
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java
@@ -139,6 +139,14 @@ public boolean updateCommitIndex(long majorityIndex, long currentTerm, boolean i
     return false;
   }
 
+  protected void updatePurgeIndex(Long purged) {
+    try (AutoCloseableLock writeLock = writeLock()) {
+      if (purged != null) {
+        purgeIndex.updateToMax(purged, infoIndexChange);
+      }
+    }
+  }
+
   protected void updateSnapshotIndexFromStateMachine() {
       updateSnapshotIndex(getSnapshotIndexFromStateMachine.getAsLong());
   }
@@ -330,9 +338,7 @@ public final CompletableFuture<Long> purge(long suggestedIndex) {
     LOG.info("{}: purge {}", getName(), suggestedIndex);
     final long finalSuggestedIndex = suggestedIndex;
     return purgeImpl(suggestedIndex).whenComplete((purged, e) -> {
-      if (purged != null) {
-        purgeIndex.updateToMax(purged, infoIndexChange);
-      }
+      updatePurgeIndex(purged);
       if (e != null) {
         LOG.warn(getName() + ": Failed to purge " + finalSuggestedIndex, e);
       }

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java
@@ -266,7 +266,7 @@ private void loadLogSegments(long lastIndexInSnapshot,
       if (!cache.isEmpty() && cache.getEndIndex() < lastIndexInSnapshot) {
         LOG.warn("End log index {} is smaller than last index in snapshot {}",
             cache.getEndIndex(), lastIndexInSnapshot);
-        purgeImpl(lastIndexInSnapshot);
+        purgeImpl(lastIndexInSnapshot).whenComplete((purged, e) -> updatePurgeIndex(purged));
       }
     }
   }
@@ -522,7 +522,7 @@ public CompletableFuture<Long> onSnapshotInstalled(long lastSnapshotIndex) {
         cacheEviction.signal();
       }
     }
-    return purgeImpl(lastSnapshotIndex);
+    return purgeImpl(lastSnapshotIndex).whenComplete((purged, e) -> updatePurgeIndex(purged));
   }
 
   @Override