Skip to content

fix: catch uncaught exception in onApply to prevent apply pipeline stall #1246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
killme2008 merged 3 commits into from
Apr 8, 2026
+243 −86
Merged

fix: catch uncaught exception in onApply to prevent apply pipeline stall #1246

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
335c17b
fix: catch uncaught exception in onApply to prevent apply pipeline stall
Qian-Cheng-nju Mar 16, 2026
fd035f3
use setError instead of setErrorAndRollback per maintainer feedback
Qian-Cheng-nju Apr 3, 2026
6a043dd
fix: rewrite flaky tests to avoid Thread.sleep and leader shutdown races
Qian-Cheng-nju Apr 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -559,7 +559,19 @@ private void doCommitted(final long committedIndex) {
}

// Apply data task to user state machine
doApplyTasks(iterImpl);
try {
doApplyTasks(iterImpl);
} catch (final Throwable t) {
// If the user state machine throws, we must not let the
// exception propagate — that would skip setLastApplied()
// and stall the apply pipeline permanently (the same
// entries would be retried on every doCommitted() call,
// hitting the same exception in an infinite loop).
LOG.error("StateMachine threw when applying entries starting at index={}. "
+ "Halting state machine to avoid stall.", iterImpl.getIndex(), t);
iterImpl.setError(new Status(RaftError.ESTATEMACHINE, "StateMachine threw: %s", t.getMessage()));
break;
}
}

if (iterImpl.hasError()) {
Expand Down
13 changes: 13 additions & 0 deletions jraft-core/src/main/java/com/alipay/sofa/jraft/core/IteratorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,19 @@ public void commitAndSnapshotSync(Closure done) {
}
}

/**
* Set error without rolling back the index. Used when the state machine
* throws an uncaught exception — we want to halt the node (ERROR state)
* without assuming any rollback policy.
*/
public void setError(final Status st) {
this.currEntry = null;
getOrCreateError().setType(EnumOutter.ErrorType.ERROR_TYPE_STATE_MACHINE);
getOrCreateError().getStatus().setError(RaftError.ESTATEMACHINE,
"StateMachine meet critical error when applying one or more tasks since index=%d, %s", this.currentIndex,
st != null ? st.toString() : "none");
}

public void setErrorAndRollback(final long ntail, final Status st) {
Requires.requireTrue(ntail > 0, "Invalid ntail=" + ntail);
if (this.currEntry == null || this.currEntry.getType() != EnumOutter.EntryType.ENTRY_TYPE_DATA) {
Expand Down
125 changes: 40 additions & 85 deletions jraft-core/src/test/java/com/alipay/sofa/jraft/core/ElectSelfPersistOrderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
package com.alipay.sofa.jraft.core;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
Expand All @@ -33,10 +32,8 @@
import com.alipay.sofa.jraft.Iterator;
import com.alipay.sofa.jraft.Node;
import com.alipay.sofa.jraft.RaftGroupService;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.entity.PeerId;
import com.alipay.sofa.jraft.entity.Task;
import com.alipay.sofa.jraft.option.NodeOptions;
import com.alipay.sofa.jraft.option.RaftMetaStorageOptions;
import com.alipay.sofa.jraft.option.RaftOptions;
Expand Down Expand Up @@ -86,16 +83,17 @@ public void teardown() throws Exception {

/**
* Verify that when setTermAndVotedFor() fails during electSelf(), the node
* steps down and does NOT send RequestVote RPCs to peers.
* steps down and does NOT become leader.
*
* Approach:
* 1. Start a 3-node cluster with a FailingMetaStorage on node 0
* 2. Wait for leader election to complete normally
* 2. Wait for leader election to complete normally (node 0 must not be leader)
* 3. Enable the failure flag on node 0's meta storage
* 4. Stop the current leader to force new elections
* 5. Node 0 will attempt electSelf(), fail on setTermAndVotedFor(), step down
* 6. Verify the observer node's term did NOT advance due to node 0's failed election
* (proving no RequestVote RPC was sent before the persistence check)
* 4. Directly trigger electSelf() on node 0 via tryElectSelf()
* 5. Verify: persist failure detected, node 0 did NOT become leader
*
* Using tryElectSelf() directly avoids the need to kill the leader (whose
* graceful shutdown sends TimeoutNow, independently bumping follower terms).
*/
@Test
public void testPersistFailurePreventsRpcSend() throws Exception {
Expand All @@ -113,9 +111,6 @@ public void testPersistFailurePreventsRpcSend() throws Exception {
for (int i = 0; i < peers.size(); i++) {
final PeerId peer = peers.get(i);
final NodeOptions nodeOptions = new NodeOptions();
// node0 gets a short election timeout so it fires first after leader death.
// Other nodes get a long timeout so they won't independently start elections
// during the test window — this lets us assert term equality on the observer.
nodeOptions.setElectionTimeoutMs(i == 0 ? 300 : 10_000);
nodeOptions.setSnapshotIntervalSecs(300);
nodeOptions.setInitialConf(conf);
Expand Down Expand Up @@ -149,91 +144,37 @@ public void onApply(final Iterator iter) {

try {
// Wait for initial leader election (fail flag is off)
Thread.sleep(3000);

Node leader = null;
for (final Node n : nodes) {
if (n.getLeaderId() != null && !n.getLeaderId().isEmpty()) {
if (n.getNodeId().getPeerId().equals(n.getLeaderId())) {
leader = n;
}
}
}
Node leader = waitForLeader(nodes, null, 30_000);
assertNotNull("Should have a leader", leader);

final Node node0 = nodes.get(0);

// If node0 is the leader, transfer leadership away
if (leader == node0) {
leader.transferLeadershipTo(peers.get(1));
Thread.sleep(2000);
leader = null;
for (final Node n : nodes) {
if (n.getLeaderId() != null && !n.getLeaderId().isEmpty()) {
if (n.getNodeId().getPeerId().equals(n.getLeaderId())) {
leader = n;
}
}
}
assumeTrue("Could not transfer leadership away from node0", leader != node0);
}

assertNotNull("Should still have a leader", leader);
final PeerId leaderPeer = leader.getLeaderId();
final int leaderIdx = peers.indexOf(leaderPeer);

// Find the observer (not node0, not leader)
Node observer = null;
int observerIdx = -1;
for (int i = 0; i < nodes.size(); i++) {
if (i != 0 && i != leaderIdx) {
observer = nodes.get(i);
observerIdx = i;
break;
}
leader = waitForLeader(nodes, node0, 30_000);
assumeTrue("Could not transfer leadership away from node0", leader != null && leader != node0);
}
assertNotNull("Should have an observer", observer);

final long observerTermBefore = ((NodeImpl) observer).getCurrentTerm();

// Enable fail flag: next setTermAndVotedFor on node0 returns false
failFlag.set(true);

// Stop leader to trigger elections
services.get(leaderIdx).shutdown();
services.get(leaderIdx).join();

// Wait for node0 to attempt election and hit the persist failure
boolean failed = failLatch.await(10, TimeUnit.SECONDS);
assertTrue("Node0 should have attempted election", failed);

final long failTerm = failedAtTerm.get();

// Brief pause for any in-flight messages
Thread.sleep(500);

final long observerTermAfter = ((NodeImpl) observer).getCurrentTerm();

// With the fix: setTermAndVotedFor is called BEFORE the RPC loop.
// When it returns false, electSelf() steps down and returns without
// sending any RequestVote RPCs. Therefore the observer's term should
// NOT have been bumped by node0's failed election attempt.
//
// Without the fix: RPCs were sent BEFORE setTermAndVotedFor, so the
// observer would have already received the RequestVote and bumped its
// term even though node0's persistence failed.
//
// The observer uses a long election timeout (10s) so it won't
// independently start its own election during this short test window.
System.out.println("Observer term before=" + observerTermBefore + ", after=" + observerTermAfter
+ ", node0 failed at term=" + failTerm);

// Observer's term must not change — proves no RequestVote RPC was sent.
assertEquals("Observer term should not advance (no RequestVote RPC sent)", observerTermBefore,
observerTermAfter);

// With the fix, node0's persist failure means it steps down and
// does NOT send RequestVote RPCs. Verify node0 is no longer leader.
// Directly trigger electSelf() on node0. With the fix, electSelf()
// calls setTermAndVotedFor() BEFORE sending RequestVote RPCs. When
// it returns false, electSelf() steps down without sending any RPCs.
// Without the fix, RPCs would be sent first and node0 could become
// leader despite the persistence failure (since quorum = 2 and the
// observer would grant the vote).
((NodeImpl) node0).tryElectSelf();

// Persist failure should have been detected
assertTrue("Node0 should have hit persist failure", failLatch.await(5, TimeUnit.SECONDS));
assertTrue("Failed term should be positive", failedAtTerm.get() > 0);

System.out.println("Node0 persist failed at term=" + failedAtTerm.get());

// Node0 must NOT be leader — proves electSelf() did not proceed
// past the failed setTermAndVotedFor() to send RPCs and win votes.
assertFalse("Node0 should not be leader after persist failure", ((NodeImpl) node0).isLeader());

} finally {
Expand All @@ -247,6 +188,20 @@ public void onApply(final Iterator iter) {
}
}

private static Node waitForLeader(List<Node> nodes, Node excluded, long timeoutMs) throws InterruptedException {
final long deadline = System.currentTimeMillis() + timeoutMs;
while (System.currentTimeMillis() < deadline) {
for (final Node n : nodes) {
if (n != excluded && n.getLeaderId() != null && !n.getLeaderId().isEmpty()
&& n.getNodeId().getPeerId().equals(n.getLeaderId())) {
return n;
}
}
Thread.sleep(100);
}
return null;
}

/**
* A RaftMetaStorage wrapper that returns false from setTermAndVotedFor()
* when the fail flag is set, simulating a disk I/O error during election.
Expand Down
Loading
Loading