You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sz...@apache.org on 2026/09/19 03:06:28 UTC
(ratis) branch master updated: RATIS-2705. Improve Division leadership API (#1600)
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 ac6bc9367 RATIS-2705. Improve Division leadership API (#1600)
ac6bc9367 is described below
commit ac6bc9367a1fb859f0c4e272c23d2195f0df2742
Author: Ivan Andika <iv...@apache.org>
AuthorDate: Sat Sep 19 11:06:21 2026 +0800
RATIS-2705. Improve Division leadership API (#1600)
Co-authored-by: szetszwo <sz...@apache.org>
---
.../java/org/apache/ratis/server/RaftServer.java | 67 +++++++--------
.../apache/ratis/server/impl/RaftServerImpl.java | 98 +++++++++++++++-------
.../org/apache/ratis/RaftExceptionBaseTest.java | 6 +-
.../ratis/server/TestRaftServerDivision.java | 54 ++----------
4 files changed, 114 insertions(+), 111 deletions(-)
diff --git a/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java b/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java
index 53760ad6f..b52a5d4e9 100644
--- a/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java
+++ b/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java
@@ -38,6 +38,7 @@ import org.apache.ratis.protocol.RaftGroupMemberId;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.LeaderNotReadyException;
+import org.apache.ratis.protocol.exceptions.LeaderSteppingDownException;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.apache.ratis.protocol.exceptions.RaftException;
import org.apache.ratis.rpc.RpcType;
@@ -65,6 +66,31 @@ public interface RaftServer extends Closeable, RpcType.Get,
interface Division extends Closeable {
Logger LOG = LoggerFactory.getLogger(Division.class);
+ enum LeadershipStatus {
+ /** The division is a leader and it is ready. */
+ LEADER_READY(null),
+ /** The division is a leader, but it is NOT ready. */
+ LEADER_NOT_READY(LeaderNotReadyException.class),
+ /** The division is a leader, but it is stepping down. */
+ LEADER_STEPPING_DOWN(LeaderSteppingDownException.class),
+ /** The division is NOT a leader; */
+ NOT_LEADER(NotLeaderException.class);
+
+ private final Class<? extends RaftException> exceptionClass;
+
+ LeadershipStatus(Class<? extends RaftException> exceptionClass) {
+ this.exceptionClass = exceptionClass;
+ }
+
+ /**
+ * @return For {@link #LEADER_READY}, return null.
+ * Otherwise, return the corresponding exception class.
+ */
+ public Class<? extends RaftException> getExceptionClass() {
+ return exceptionClass;
+ }
+ }
+
/** @return the {@link DivisionProperties} for this division. */
DivisionProperties properties();
@@ -85,45 +111,16 @@ public interface RaftServer extends Closeable, RpcType.Get,
/** @return the information about this division. */
DivisionInfo getInfo();
- /**
- * Check whether this division is the leader and ready.
- *
- * @return null if this division is the leader and ready; otherwise,
- * return a {@link NotLeaderException} or a
- * {@link LeaderNotReadyException}.
- */
- default RaftException checkLeaderReady() {
- if (!getInfo().isLeader()) {
- return newNotLeaderException();
- }
- return getInfo().isLeaderReady() ? null
- : new LeaderNotReadyException(getMemberId());
- }
+ /** @return the current leadership status of this division. */
+ LeadershipStatus getCurrentLeadershipStatus();
/**
- * Create a {@link NotLeaderException} using the current division state.
+ * Convert the given leadership status (not necessarily the current status) to an exception.
*
- * <p>The suggested leader and peers are best-effort hints. The suggested
- * leader is null if this division is not running or if its current leader
- * ID is unknown or is the ID of this division.</p>
- *
- * <p>This method always creates a {@link NotLeaderException}; it does not
- * check whether this division is the leader or ready to serve requests.
- * Use {@link #checkLeaderReady()} when an admission check is required.</p>
+ * @return For {@link LeadershipStatus#LEADER_READY}, return null.
+ * Otherwise, return an exception corresponding to {@link LeadershipStatus#getExceptionClass()}.
*/
- default NotLeaderException newNotLeaderException() {
- if (!getInfo().getLifeCycleState().isRunning()) {
- return new NotLeaderException(getMemberId(), null, null);
- }
- RaftPeerId leaderId = getInfo().getLeaderId();
- if (leaderId == null || leaderId.equals(getId())) {
- // No idea about who is the current leader. Or the peer is the current
- // leader, but it is about to step down. set the suggested leader as null.
- leaderId = null;
- }
- final RaftConfiguration conf = getRaftConf();
- return new NotLeaderException(getMemberId(), conf.getPeer(leaderId), conf.getAllPeers());
- }
+ RaftException newLeadershipException(LeadershipStatus leadershipStatus);
/** @return the {@link RaftGroup} for this division. */
default RaftGroup getGroup() {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index ff2723904..c5df90dc5 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -78,7 +78,9 @@ import org.apache.ratis.protocol.exceptions.TransferLeadershipException;
import org.apache.ratis.server.DataStreamMap;
import org.apache.ratis.server.DivisionInfo;
import org.apache.ratis.server.DivisionProperties;
+import org.apache.ratis.server.RaftConfiguration;
import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.RaftServer.Division.LeadershipStatus;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.RaftServerRpc;
import org.apache.ratis.server.impl.LeaderElection.Phase;
@@ -781,28 +783,64 @@ class RaftServerImpl implements RaftServer.Division,
*/
private CompletableFuture<RaftClientReply> checkLeaderState(
RaftClientRequest request, CacheEntry entry, TransactionContextImpl context) {
- if (!getInfo().isLeader()) {
- NotLeaderException exception = newNotLeaderException();
- final RaftClientReply reply = newExceptionReply(request, exception);
- return failWithReply(reply, entry, context);
- }
- if (!getInfo().isLeaderReady()) {
- final CacheEntry cacheEntry = retryCache.getIfPresent(ClientInvocationId.valueOf(request));
- if (cacheEntry != null && cacheEntry.isCompletedNormally()) {
- return cacheEntry.getReplyFuture();
- }
- final LeaderNotReadyException lnre = new LeaderNotReadyException(getMemberId());
- final RaftClientReply reply = newExceptionReply(request, lnre);
- return failWithReply(reply, entry, context);
+ final LeadershipStatus leadershipStatus = getCurrentLeadershipStatus();
+ switch (leadershipStatus) {
+ case NOT_LEADER:
+ return failWithReply(request, leadershipStatus, entry, context);
+ case LEADER_NOT_READY:
+ final CacheEntry cacheEntry = retryCache.getIfPresent(ClientInvocationId.valueOf(request));
+ if (cacheEntry != null && cacheEntry.isCompletedNormally()) {
+ return cacheEntry.getReplyFuture();
+ }
+ return failWithReply(request, leadershipStatus, entry, context);
+ case LEADER_STEPPING_DOWN:
+ if (!request.isReadOnly()) {
+ return failWithReply(request, leadershipStatus, entry, context);
+ }
+ return null;
+ case LEADER_READY:
+ return null;
+ default:
+ throw new IllegalStateException("Unexpected LeadershipStatus: " + leadershipStatus);
}
+ }
- if (!request.isReadOnly() && isSteppingDown()) {
- final LeaderSteppingDownException lsde = new LeaderSteppingDownException(getMemberId() + " is stepping down");
- final RaftClientReply reply = newExceptionReply(request, lsde);
- return failWithReply(reply, entry, context);
+ @Override
+ public LeadershipStatus getCurrentLeadershipStatus() {
+ return !getInfo().isLeader() ? LeadershipStatus.NOT_LEADER
+ : !getInfo().isLeaderReady() ? LeadershipStatus.LEADER_NOT_READY
+ : isSteppingDown() ? LeadershipStatus.LEADER_STEPPING_DOWN
+ : LeadershipStatus.LEADER_READY;
+ }
+
+ @Override
+ public RaftException newLeadershipException(LeadershipStatus leadershipStatus) {
+ switch (leadershipStatus) {
+ case NOT_LEADER:
+ return newNotLeaderException();
+ case LEADER_NOT_READY:
+ return new LeaderNotReadyException(getMemberId());
+ case LEADER_STEPPING_DOWN:
+ return new LeaderSteppingDownException(getMemberId() + " is stepping down");
+ case LEADER_READY:
+ return null;
+ default:
+ throw new IllegalStateException("Unexpected LeadershipStatus: " + leadershipStatus);
}
+ }
- return null;
+ NotLeaderException newNotLeaderException() {
+ if (!getInfo().getLifeCycleState().isRunning()) {
+ return new NotLeaderException(getMemberId(), null, null);
+ }
+ RaftPeerId leaderId = getInfo().getLeaderId();
+ if (leaderId == null || leaderId.equals(getId())) {
+ // No idea about who is the current leader. Or the peer is the current
+ // leader, but it is about to step down. set the suggested leader as null.
+ leaderId = null;
+ }
+ final RaftConfiguration conf = getRaftConf();
+ return new NotLeaderException(getMemberId(), conf.getPeer(leaderId), conf.getAllPeers());
}
void assertLifeCycleState(Set<LifeCycle.State> expected) throws ServerNotReadyException {
@@ -819,11 +857,17 @@ class RaftServerImpl implements RaftServer.Division,
}
private CompletableFuture<RaftClientReply> failWithReply(
- RaftClientReply reply, CacheEntry entry, TransactionContextImpl context) {
+ RaftClientRequest request, LeadershipStatus leadershipStatus, CacheEntry entry, TransactionContextImpl context) {
+ return failWithReply(request, newLeadershipException(leadershipStatus), entry, context);
+ }
+
+ private CompletableFuture<RaftClientReply> failWithReply(
+ RaftClientRequest request, RaftException e, CacheEntry entry, TransactionContextImpl context) {
if (context != null) {
- cancelTransaction(context, reply.getException());
+ cancelTransaction(context, e);
}
+ final RaftClientReply reply = newExceptionReply(request, e);
if (entry == null) {
return CompletableFuture.completedFuture(reply);
}
@@ -860,9 +904,7 @@ class RaftServerImpl implements RaftServer.Division,
final LeaderStateImpl unsyncedLeaderState = role.getLeaderState().orElse(null);
if (unsyncedLeaderState == null) {
- final NotLeaderException nle = newNotLeaderException();
- final RaftClientReply reply = newExceptionReply(request, nle);
- return failWithReply(reply, cacheEntry, context);
+ return failWithReply(request, LeadershipStatus.NOT_LEADER, cacheEntry, context);
}
final PendingRequests.Permit unsyncedPermit = unsyncedLeaderState.tryAcquirePendingRequest(request.getMessage());
if (unsyncedPermit == null) {
@@ -889,14 +931,13 @@ class RaftServerImpl implements RaftServer.Division,
try {
state.appendLog(context);
} catch (StateMachineException e) {
- // the StateMachineException is thrown by the SM in the preAppend stage.
- // Return the exception in a RaftClientReply.
- final RaftClientReply exceptionReply = newExceptionReply(request, e);
// leader will step down here
if (e.leaderShouldStepDown() && getInfo().isLeader()) {
leaderState.submitStepDownEvent(StepDownReason.STATE_MACHINE_EXCEPTION);
}
- return failWithReply(exceptionReply, cacheEntry, null);
+ // the StateMachineException is thrown by the SM in the preAppend stage.
+ // Return the exception in a RaftClientReply.
+ return failWithReply(request, e, cacheEntry, null);
}
// put the request into the pending queue
@@ -1041,8 +1082,7 @@ class RaftServerImpl implements RaftServer.Division,
if (context.getException() != null) {
final Exception exception = context.getException();
final StateMachineException e = new StateMachineException(getMemberId(), exception);
- final RaftClientReply exceptionReply = newExceptionReply(request, e);
- return failWithReply(exceptionReply, cacheEntry, context);
+ return failWithReply(request, e, cacheEntry, context);
}
try {
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftExceptionBaseTest.java b/ratis-server/src/test/java/org/apache/ratis/RaftExceptionBaseTest.java
index eba2e27ec..251fec74d 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftExceptionBaseTest.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftExceptionBaseTest.java
@@ -101,14 +101,16 @@ public abstract class RaftExceptionBaseTest<CLUSTER extends MiniRaftCluster>
public void testNewNotLeaderException() throws Exception {
runWithNewCluster(NUM_PEERS, cluster -> {
final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
- final NotLeaderException leaderException = leader.newNotLeaderException();
+ final NotLeaderException leaderException = (NotLeaderException) leader.newLeadershipException(
+ RaftServer.Division.LeadershipStatus.NOT_LEADER);
Assertions.assertNull(leaderException.getSuggestedLeader());
Assertions.assertEquals(cluster.getPeers().size(), leaderException.getPeers().size());
Assertions.assertTrue(cluster.getPeers().containsAll(leaderException.getPeers()));
final RaftServer.Division follower = cluster.getFollowers().get(0);
follower.close();
- final NotLeaderException closedException = follower.newNotLeaderException();
+ final NotLeaderException closedException = (NotLeaderException) follower.newLeadershipException(
+ RaftServer.Division.LeadershipStatus.NOT_LEADER);
Assertions.assertNull(closedException.getSuggestedLeader());
Assertions.assertTrue(closedException.getPeers().isEmpty());
});
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerDivision.java b/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerDivision.java
index 9257bad2f..99a58b42b 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerDivision.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerDivision.java
@@ -18,57 +18,21 @@
*/
package org.apache.ratis.server;
-import java.util.Arrays;
-import org.apache.ratis.protocol.RaftGroupId;
-import org.apache.ratis.protocol.RaftGroupMemberId;
-import org.apache.ratis.protocol.RaftPeer;
-import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.LeaderNotReadyException;
+import org.apache.ratis.protocol.exceptions.LeaderSteppingDownException;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
-import org.apache.ratis.protocol.exceptions.RaftException;
-import org.apache.ratis.util.LifeCycle;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import static org.mockito.Answers.CALLS_REAL_METHODS;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
class TestRaftServerDivision {
- private static final RaftPeerId LOCAL_ID = RaftPeerId.valueOf("local");
- private static final RaftPeerId REMOTE_ID = RaftPeerId.valueOf("remote");
- private static final RaftGroupMemberId MEMBER_ID =
- RaftGroupMemberId.valueOf(LOCAL_ID, RaftGroupId.emptyGroupId());
- private static final RaftPeer REMOTE_PEER = RaftPeer.newBuilder().setId(REMOTE_ID).build();
-
@Test
- void testCheckLeaderReady() {
- final RaftServer.Division division = mock(RaftServer.Division.class, CALLS_REAL_METHODS);
- final DivisionInfo info = mock(DivisionInfo.class);
- final RaftConfiguration conf = mock(RaftConfiguration.class);
- when(division.getMemberId()).thenReturn(MEMBER_ID);
- when(division.getInfo()).thenReturn(info);
- when(division.getRaftConf()).thenReturn(conf);
- when(info.getLifeCycleState()).thenReturn(LifeCycle.State.RUNNING);
- when(conf.getAllPeers()).thenReturn(Arrays.asList(RaftPeer.newBuilder().setId(LOCAL_ID).build(), REMOTE_PEER));
-
- when(info.isLeader()).thenReturn(true);
- when(info.isLeaderReady()).thenReturn(true);
- Assertions.assertNull(division.checkLeaderReady());
-
- when(info.isLeaderReady()).thenReturn(false);
- Assertions.assertInstanceOf(LeaderNotReadyException.class, division.checkLeaderReady());
-
- when(info.isLeader()).thenReturn(false);
- when(info.getLeaderId()).thenReturn(LOCAL_ID);
- final RaftException staleLeader = division.checkLeaderReady();
- Assertions.assertInstanceOf(NotLeaderException.class, staleLeader);
- Assertions.assertNull(((NotLeaderException) staleLeader).getSuggestedLeader());
-
- when(info.getLeaderId()).thenReturn(REMOTE_ID);
- when(conf.getPeer(REMOTE_ID)).thenReturn(REMOTE_PEER);
- final RaftException knownLeader = division.checkLeaderReady();
- Assertions.assertInstanceOf(NotLeaderException.class, knownLeader);
- Assertions.assertEquals(REMOTE_PEER, ((NotLeaderException) knownLeader).getSuggestedLeader());
+ void testLeadershipStatusExceptionClasses() {
+ Assertions.assertNull(RaftServer.Division.LeadershipStatus.LEADER_READY.getExceptionClass());
+ Assertions.assertEquals(LeaderNotReadyException.class,
+ RaftServer.Division.LeadershipStatus.LEADER_NOT_READY.getExceptionClass());
+ Assertions.assertEquals(LeaderSteppingDownException.class,
+ RaftServer.Division.LeadershipStatus.LEADER_STEPPING_DOWN.getExceptionClass());
+ Assertions.assertEquals(NotLeaderException.class,
+ RaftServer.Division.LeadershipStatus.NOT_LEADER.getExceptionClass());
}
}