(celeborn) branch branch-0.7 updated: [CELEBORN-2453] Master should fail fast when the Ratis server is closed unexpectedly

[za...@apache.org]

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

zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/branch-0.7 by this push:
     new 493252df0d [CELEBORN-2453] Master should fail fast when the Ratis server is closed unexpectedly
493252df0d is described below

commit 493252df0d9fd8c806d20d177226fe2530035bb4
Author: yew1eb <ye...@gmail.com>
AuthorDate: Sat Oct 10 16:32:34 2026 +0800

    [CELEBORN-2453] Master should fail fast when the Ratis server is closed unexpectedly
    
    ### What changes were proposed in this pull request?
    When the embedded Ratis server is closed unexpectedly (e.g. by `JvmPauseMonitor` after a JVM pause longer than `raft.server.close.threshold`), the master process keeps running as a zombie: its RPC/HTTP ports stay alive but it has silently left the Raft group.
    This PR makes the master fail fast in this situation:
    1. Add a new config `celeborn.master.ha.ratis.unexpected.close.action` (`exit` / `none`, default `exit`). When the periodic role check detects the local `RaftServer` has reached a terminal lifecycle state (`CLOSED` / `EXCEPTION`), the master logs the root cause and exits, so the HA group keeps a healthy quorum and the node can be restarted by external supervision (systemd/k8s).
    2. `HARaftServer.updateServerRole()` short-circuits when the Raft server is in a terminal state (`CLOSED` / `EXCEPTION`), avoiding repeated ERROR log flooding from subsequent role checks and metrics scrapes, while leaving transitional states (`STARTING` / `CLOSING`) on the existing code path.
    
    ### Why are the changes needed?
     In our production, one master of a 3-node HA cluster got stuck for ~78s by `jmap -F -histo` (non-GC pause). Ratis `JvmPauseMonitor` then closed the local RaftServer:
    <img width="2314" height="1124" alt="image" src="https://github.com/user-attachments/assets/1ed38a5b-e707-4a80-83fe-4aa9f802700d" />
    
    After that the master process was still alive but out of the Raft group, role became `UNRECOGNIZED`, and every metrics scrape / role check printed a full ERROR stack (`ServerNotReadyException`, `SegmentedRaftLog is expected to be opened but it is CLOSED`). Nothing recovers the node except a manual restart — failing fast makes the failure explicit and lets supervision restart it.
    
    ### Does this PR resolve a correctness bug?
    
    - [ ] Yes
    
    ### Does this PR introduce _any_ user-facing change?
    
    - [ ] Yes
    
    ### How was this patch tested?
    - New unit test `RatisMasterStatusSystemSuiteJ#testUnexpectedRaftServerClose`.
    - Existing master module tests pass (`./build/mvn test -pl master`).
    
    This patch had conflicts when merged, resolved by
    Committer: zhengtao <sh...@alibaba-inc.com>
    
    Closes #3838 from yew1eb/CELEBORN-2453.
    
    Authored-by: yew1eb <ye...@gmail.com>
    Signed-off-by: zhengtao <sh...@alibaba-inc.com>
    
    AI-Contributed/Feature: 0/76
    AI-Contributed/UT: 0/58
    
    (cherry picked from commit abdfd05d804e8a128682d1d87c540501ed5f8be2)
    Signed-off-by: zhengtao <sh...@alibaba-inc.com>
---
 .../org/apache/celeborn/common/CelebornConf.scala  | 14 +++++
 docs/configuration/ha.md                           |  1 +
 .../deploy/master/clustermeta/ha/HARaftServer.java | 61 ++++++++++++++++++++++
 .../ha/RatisMasterStatusSystemSuiteJ.java          | 58 ++++++++++++++++++++
 4 files changed, 134 insertions(+)

diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 71f9b3d01d..1fadc77f23 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -811,6 +811,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
     get(HA_MASTER_RATIS_NOTIFICATION_NO_LEADER_TIMEOUT)
   def haMasterRatisRpcSlownessTimeout: Long = get(HA_MASTER_RATIS_RPC_SLOWNESS_TIMEOUT)
   def haMasterRatisRoleCheckInterval: Long = get(HA_MASTER_RATIS_ROLE_CHECK_INTERVAL)
+  def haMasterRatisUnexpectedCloseAction: String = get(HA_MASTER_RATIS_UNEXPECTED_CLOSE_ACTION)
   def haMasterRatisSnapshotAutoTriggerEnabled: Boolean =
     get(HA_MASTER_RATIS_SNAPSHOT_AUTO_TRIGGER_ENABLED)
   def haMasterRatisSnapshotAutoTriggerThreshold: Long =
@@ -2976,6 +2977,19 @@ object CelebornConf extends Logging {
       .timeConf(TimeUnit.SECONDS)
       .createWithDefaultString("60s")
 
+  val HA_MASTER_RATIS_UNEXPECTED_CLOSE_ACTION: ConfigEntry[String] =
+    buildConf("celeborn.master.ha.ratis.unexpected.close.action")
+      .categories("ha")
+      .version("0.7.1")
+      .doc("Action when the local raft server is closed unexpectedly, i.e. not via the " +
+        "master stop flow, e.g. closed by the Ratis JvmPauseMonitor after a JVM pause " +
+        "longer than raft.server.close.threshold. 'exit' logs a fatal message and exits " +
+        "the master process so that supervisors (systemd/K8s) can restart it and the " +
+        "node rejoins the raft group; 'none' only logs.")
+      .stringConf
+      .checkValues(Set("exit", "none"))
+      .createWithDefault("exit")
+
   val HA_MASTER_RATIS_SERVER_RETRY_CACHE_EXPIRY_TIME: ConfigEntry[Long] =
     buildConf("celeborn.master.ha.ratis.raft.server.retrycache.expirytime")
       .withAlternative("celeborn.ha.master.ratis.raft.server.retrycache.expirytime")
diff --git a/docs/configuration/ha.md b/docs/configuration/ha.md
index a249752fd6..0e2857951d 100644
--- a/docs/configuration/ha.md
+++ b/docs/configuration/ha.md
@@ -30,4 +30,5 @@ license: |
 | celeborn.master.ha.ratis.raft.server.close.threshold | 60s | false | Threshold for the Ratis JvmPauseMonitor to close the local raft server after a JVM/host pause, e.g. long GC, VM live migration or jmap -F. Maps to Ratis raft.server.close.threshold. Once exceeded, the master leaves the raft group until the process is restarted. Takes precedence over the celeborn.ratis.raft.server.close.threshold passthrough when both are set. | 0.7.1 |  |
 | celeborn.master.ha.ratis.raft.server.storage.dir | /tmp/ratis | false | Root storage directory to hold RaftServer data. | 0.3.0 | celeborn.ha.master.ratis.raft.server.storage.dir | 
 | celeborn.master.ha.ratis.raft.server.storage.startup.option | RECOVER | false | Startup option of RaftServer storage. Available options: RECOVER, FORMAT. | 0.5.0 |  | 
+| celeborn.master.ha.ratis.unexpected.close.action | exit | false | Action when the local raft server is closed unexpectedly, i.e. not via the master stop flow, e.g. closed by the Ratis JvmPauseMonitor after a JVM pause longer than raft.server.close.threshold. 'exit' logs a fatal message and exits the master process so that supervisors (systemd/K8s) can restart it and the node rejoins the raft group; 'none' only logs. | 0.7.1 |  |
 <!--end-include-->
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java
index 597d6bf46a..c36aa321de 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java
@@ -101,9 +101,16 @@ public class HARaftServer {
 
   private final CelebornConf conf;
   private final AtomicBoolean stopped = new AtomicBoolean(false);
+  private final AtomicBoolean unexpectedCloseHandled = new AtomicBoolean(false);
   private long workerTimeoutDeadline;
   private long appTimeoutDeadline;
 
+  private Runnable unexpectedCloseHandler =
+      () -> {
+        LOG.error("Exiting master process because the raft server was closed unexpectedly.");
+        System.exit(1);
+      };
+
   /**
    * Returns a Master Ratis server.
    *
@@ -158,6 +165,7 @@ public class HARaftServer {
     // Run a scheduler to check and update the server role on the leader periodically
     this.scheduledRoleChecker.scheduleWithFixedDelay(
         () -> {
+          checkRaftServerState();
           // Run this check only on the leader OM
           if (cachedPeerRole.isPresent()
               && cachedPeerRole.get() == RaftProtos.RaftPeerRole.LEADER) {
@@ -311,6 +319,46 @@ public class HARaftServer {
     }
   }
 
+  @VisibleForTesting
+  void checkRaftServerState() {
+    if (stopped.get() || unexpectedCloseHandled.get()) {
+      return;
+    }
+    handleRaftServerState(server.getLifeCycleState());
+  }
+
+  @VisibleForTesting
+  void handleRaftServerState(LifeCycle.State state) {
+    // Recheck stopped after the lifecycle state was observed: a concurrent stop() may
+    // have set stopped and closed the server in between. Treating an intentional
+    // shutdown as an unexpected close would exit the process as a failure and trigger
+    // a spurious restart under a restart-on-failure supervisor.
+    if (stopped.get() || unexpectedCloseHandled.get()) {
+      return;
+    }
+    if ((state == LifeCycle.State.CLOSED || state == LifeCycle.State.EXCEPTION)
+        && unexpectedCloseHandled.compareAndSet(false, true)) {
+      LOG.error(
+          "Raft server {} was closed unexpectedly (state {}), e.g. by the Ratis "
+              + "JvmPauseMonitor after a long JVM pause. Action: {}.",
+          server.getId(),
+          state,
+          conf.haMasterRatisUnexpectedCloseAction());
+      onUnexpectedRaftServerClose();
+    }
+  }
+
+  void onUnexpectedRaftServerClose() {
+    if ("exit".equals(conf.haMasterRatisUnexpectedCloseAction())) {
+      unexpectedCloseHandler.run();
+    }
+  }
+
+  @VisibleForTesting
+  void setUnexpectedCloseHandler(Runnable handler) {
+    this.unexpectedCloseHandler = handler;
+  }
+
   private RaftProperties newRaftProperties(CelebornConf conf, RpcType rpc) {
     final RaftProperties properties = new RaftProperties();
     // Set RPC type
@@ -557,6 +605,19 @@ public class HARaftServer {
    * Get the group info (peer role and leader peer id) from Ratis server and update the server role.
    */
   public void updateServerRole() {
+    LifeCycle.State state = server.getLifeCycleState();
+    if (state != LifeCycle.State.RUNNING) {
+      // Only log on role change to avoid flooding logs on every metrics scrape.
+      if (cachedPeerRole.isPresent()) {
+        LOG.warn(
+            "Raft server {} is not RUNNING (state {}). Setting cached role to UNRECOGNIZED "
+                + "and resetting leader info.",
+            server.getId(),
+            state);
+      }
+      setServerRole(null, null, null);
+      return;
+    }
     try {
       GroupInfoReply groupInfo = getGroupInfo();
       RaftProtos.RoleInfoProto roleInfoProto = groupInfo.getRoleInfoProto();
diff --git a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
index 6038ee4e15..7ca8a83cf9 100644
--- a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
+++ b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
@@ -25,11 +25,13 @@ import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 import scala.Tuple2;
 
 import com.google.common.collect.Lists;
+import org.apache.ratis.util.LifeCycle;
 import org.junit.*;
 import org.mockito.Mockito;
 
@@ -2145,4 +2147,60 @@ public class RatisMasterStatusSystemSuiteJ {
     assertEquals(statusSystem.applicationInfos.get(appId).userIdentifier(), userIdentifier);
     assertEquals(statusSystem.applicationInfos.get(appId).extraInfo(), extraInfo);
   }
+
+  @Test
+  public void testUnexpectedRaftServerClose() throws Exception {
+    HARaftServer nonLeader =
+        !RATISSERVER1.isLeader()
+            ? RATISSERVER1
+            : (!RATISSERVER2.isLeader() ? RATISSERVER2 : RATISSERVER3);
+    try {
+      AtomicBoolean handlerInvoked = new AtomicBoolean(false);
+      nonLeader.setUnexpectedCloseHandler(() -> handlerInvoked.set(true));
+
+      // Simulate the Ratis JvmPauseMonitor closing the local raft server.
+      nonLeader.getServer().close();
+      Assert.assertEquals(LifeCycle.State.CLOSED, nonLeader.getServerState());
+
+      long deadline = System.currentTimeMillis() + 30_000;
+      while (!handlerInvoked.get() && System.currentTimeMillis() < deadline) {
+        Thread.sleep(200L);
+      }
+      Assert.assertTrue(handlerInvoked.get());
+
+      nonLeader.updateServerRole();
+      Assert.assertFalse(nonLeader.getCachedLeaderPeerRpcEndpoint().isPresent());
+    } finally {
+      init();
+    }
+  }
+
+  @Test
+  public void testConcurrentStopNotTreatedAsUnexpectedClose() throws Exception {
+    HARaftServer nonLeader =
+        !RATISSERVER1.isLeader()
+            ? RATISSERVER1
+            : (!RATISSERVER2.isLeader() ? RATISSERVER2 : RATISSERVER3);
+    try {
+      AtomicBoolean handlerInvoked = new AtomicBoolean(false);
+      nonLeader.setUnexpectedCloseHandler(() -> handlerInvoked.set(true));
+
+      // Normal shutdown: sets stopped and closes the raft server.
+      nonLeader.stop();
+      Assert.assertEquals(LifeCycle.State.CLOSED, nonLeader.getServerState());
+
+      // Replay the interleaving described in the review: the checker passed the entry
+      // check while stopped was still false and read CLOSED; stop() then completed
+      // before the checker acted on the state. The checker must stay silent.
+      nonLeader.handleRaftServerState(LifeCycle.State.CLOSED);
+      Assert.assertFalse(
+          "A normal stop must not be treated as an unexpected close", handlerInvoked.get());
+
+      // The full check path must also stay silent after a normal stop.
+      nonLeader.checkRaftServerState();
+      Assert.assertFalse(handlerInvoked.get());
+    } finally {
+      init();
+    }
+  }
 }