(ratis) branch master updated: RATIS-2723. LogAppender purge tests may fail after a leader change. (#1619)

[sz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git


The following commit(s) were added to refs/heads/master by this push:
     new dd518d91e RATIS-2723. LogAppender purge tests may fail after a leader change. (#1619)
dd518d91e is described below

commit dd518d91e74f381dff894b3c13a3fc68a5a914a9
Author: slfan1989 <55...@users.noreply.github.com>
AuthorDate: Fri Oct 2 00:56:57 2026 +0800

    RATIS-2723. LogAppender purge tests may fail after a leader change. (#1619)
---
 .../java/org/apache/ratis/LogAppenderTests.java    | 64 ++++++++++++++++------
 1 file changed, 46 insertions(+), 18 deletions(-)

diff --git a/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java b/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java
index 8a8731daf..77b23a8b8 100644
--- a/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java
@@ -231,11 +231,8 @@ public abstract class LogAppenderTests<CLUSTER extends MiniRaftCluster>
     final RaftProperties prop = getProperties();
     RaftServerConfigKeys.Log.setPurgeGap(prop, 1);
     RaftServerConfigKeys.Log.setSegmentSizeMax(prop, SizeInBytes.valueOf("1KB"));
-    runWithNewCluster(3, cluster -> {
-      final long startIndexAfterPurge = setupPurgedLeaderLog(cluster);
-      // Test when followerNextIndex < leader's logStartIndex
-      runTestNewAppendEntriesRequestAfterPurge(cluster, startIndexAfterPurge - 1);
-    });
+    // Test when followerNextIndex < leader's logStartIndex.
+    runWithNewCluster(3, cluster -> runTestNewAppendEntriesRequestAfterPurge(cluster, true));
   }
 
   @Test
@@ -243,23 +240,55 @@ public abstract class LogAppenderTests<CLUSTER extends MiniRaftCluster>
     final RaftProperties prop = getProperties();
     RaftServerConfigKeys.Log.setPurgeGap(prop, 1);
     RaftServerConfigKeys.Log.setSegmentSizeMax(prop, SizeInBytes.valueOf("1KB"));
-    runWithNewCluster(3, cluster -> {
-      final long startIndexAfterPurge = setupPurgedLeaderLog(cluster);
-      // Test when followerNextIndex == leader's logStartIndex, but the previous index is already purged
-      runTestNewAppendEntriesRequestAfterPurge(cluster, startIndexAfterPurge);
-    });
+    // Test when followerNextIndex == leader's logStartIndex, but the previous index is already purged.
+    runWithNewCluster(3, cluster -> runTestNewAppendEntriesRequestAfterPurge(cluster, false));
   }
 
-  private long setupPurgedLeaderLog(CLUSTER cluster) throws Exception {
-    final RaftServer.Division leader = waitForLeader(cluster);
-    final RaftLog leaderLog = leader.getRaftLog();
+  private void runTestNewAppendEntriesRequestAfterPurge(CLUSTER cluster,
+      boolean followerBehindStartIndex) throws Exception {
+    final int maxAttempts = 3;
+    for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+      try (RaftClient client = cluster.createClient(waitForLeader(cluster).getId())) {
+        for (SimpleMessage msg : generateMsgs(5)) {
+          client.io().send(msg);
+        }
+      }
 
-    try (RaftClient client = cluster.createClient(leader.getId())) {
-      for (SimpleMessage msg : generateMsgs(5)) {
-        client.io().send(msg);
+      // Sending messages may change the leader. Use the same leader and term for purge and verification.
+      final RaftServer.Division leader = waitForLeader(cluster);
+      final long term = leader.getInfo().getCurrentTerm();
+      try {
+        final long startIndexAfterPurge = setupPurgedLeaderLog(leader);
+        // Verify only if the node is still the leader in the term used for purge.
+        if (isLeaderInTerm(leader, term)) {
+          runTestNewAppendEntriesRequestAfterPurge(leader,
+              followerBehindStartIndex ? startIndexAfterPurge - 1 : startIndexAfterPurge);
+          // Leadership may change during verification; accept the result only if it is still valid.
+          if (isLeaderInTerm(leader, term)) {
+            return;
+          }
+        }
+      } catch (InterruptedException e) {
+        throw e;
+      } catch (Exception | AssertionError e) {
+        if (isLeaderInTerm(leader, term)) {
+          throw e;
+        }
+        LOG.info("Leader {} changed during purge verification in term {}", leader.getId(), term, e);
       }
+      LOG.info("Leader {} changed in term {} during purge test (attempt {}/{})",
+          leader.getId(), term, attempt, maxAttempts);
     }
+    Assertions.fail("Leader changed during all " + maxAttempts + " purge test attempts");
+  }
 
+  // Check both role and term: the same server may step down and become leader again in a later term.
+  private static boolean isLeaderInTerm(RaftServer.Division leader, long term) {
+    return leader.getInfo().isLeader() && leader.getInfo().getCurrentTerm() == term;
+  }
+
+  private long setupPurgedLeaderLog(RaftServer.Division leader) throws Exception {
+    final RaftLog leaderLog = leader.getRaftLog();
     final long lastLogIndex = leaderLog.getLastEntryTermIndex().getIndex();
     LOG.info("Leader log lastIndex={}, startIndex={}", lastLogIndex, leaderLog.getStartIndex());
     Assertions.assertTrue(lastLogIndex > 5, "Need enough log entries for the test");
@@ -281,9 +310,8 @@ public abstract class LogAppenderTests<CLUSTER extends MiniRaftCluster>
     return startIndexAfterPurge;
   }
 
-  void runTestNewAppendEntriesRequestAfterPurge(CLUSTER cluster,
+  void runTestNewAppendEntriesRequestAfterPurge(RaftServer.Division leader,
       long targetNextIndex) throws Exception {
-    final RaftServer.Division leader = waitForLeader(cluster);
     final RaftLog leaderLog = leader.getRaftLog();
     final long startIndexAfterPurge = leaderLog.getStartIndex();