Copilot commented on code in PR #3838:
URL: https://github.com/apache/celeborn/pull/3838#discussion_r3967204123
##########
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);
+ };
+
Review Comment:
Calling `System.exit(1)` from inside `HARaftServer` makes shutdown behavior
harder to control (e.g., it can bypass any graceful master shutdown flow and
make tests/in-process embedding brittle). Consider delegating the termination
decision to the master’s top-level lifecycle (e.g., invoking a master shutdown
callback or throwing a fatal error that is handled at the process entrypoint),
while keeping the default behavior functionally equivalent. If keeping
`System.exit`, it would also be helpful to include more diagnostic context in
the log (e.g., a cause/exception if available) since the description mentions
logging the root cause.
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java:
##########
@@ -311,6 +319,34 @@ public void stop(boolean transferLeadership) {
}
}
+ private void checkRaftServerState() {
+ if (stopped.get() || unexpectedCloseHandled.get()) {
+ return;
+ }
+ LifeCycle.State state = server.getLifeCycleState();
+ 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();
+ }
+ }
Review Comment:
The PR description says the periodic role check should react when the local
RaftServer is no longer RUNNING (and the node is still in the Raft group). The
current implementation triggers only for CLOSED/EXCEPTION and does not check
group membership. Please align implementation and description: either (a)
broaden the condition to `state != RUNNING` (if that’s intended), and/or (b)
explicitly verify the local peer is still a member of the group before taking
the configured action; otherwise update the description/docs to match the
narrower behavior.
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java:
##########
@@ -540,6 +576,19 @@ public Optional<LeaderPeerEndpoints>
getCachedLeaderPeerRpcEndpoint() {
* 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;
+ }
Review Comment:
This change resets the cached role/leader info for *any* non-RUNNING
lifecycle state. That includes transitional states (e.g., STARTING/STOPPING,
depending on Ratis lifecycle), which can cause metrics/role to flap to
UNRECOGNIZED during normal transitions rather than only when the server is
effectively unusable. To keep the behavior targeted to the log-flooding
scenario, consider narrowing this guard to terminal/failure states (e.g.,
CLOSED/EXCEPTION) or gating it behind `unexpectedCloseHandled` so transitional
states don’t unnecessarily clear role/leader info.
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HARaftServer.java:
##########
@@ -311,6 +319,34 @@ public void stop(boolean transferLeadership) {
}
}
+ private void checkRaftServerState() {
+ if (stopped.get() || unexpectedCloseHandled.get()) {
+ return;
+ }
+ LifeCycle.State state = server.getLifeCycleState();
+ 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;
Review Comment:
`setUnexpectedCloseHandler` allows setting a null handler, which would lead
to a null dereference when an unexpected close occurs. Since this is a test
hook, it’s low cost to harden it by rejecting null (e.g., via a precondition
check) to avoid future accidental NPEs in tests.
##########
master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java:
##########
@@ -2145,4 +2147,31 @@ public void testRegisterApplicationInfo() {
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());
Review Comment:
This test relies on a fixed polling loop and real sleeps to observe the
scheduled role checker invoking the handler, which can be flaky and slow under
CI load. Prefer using a dedicated async waiting utility (e.g., an
‘eventually’/Awaitility-style wait) and/or make the triggering deterministic
for tests (for example, by exposing a `@VisibleForTesting` method to run the
state check once, or configuring the role-check interval to be very small
within this suite) to reduce timing sensitivity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
