Skip to content

feat: adds EventBus and introduces Mpsc as alternative of disruptor #1238

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

Draft
killme2008 wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

feat: adds EventBus and introduces Mpsc as alternative of disruptor #1238

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bb4715e
feat: adds EventBus and introduces Mpsc as alternative of disruptor
killme2008 Dec 18, 2025
fc95cd1
feat: make EventBus to be configurable and as SPI
killme2008 Dec 23, 2025
7960158
fix: test
killme2008 Dec 24, 2025
03afaf8
fix: shutdown
killme2008 Dec 24, 2025
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
151 changes: 57 additions & 94 deletions 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 @@ -47,21 +47,14 @@
import com.alipay.sofa.jraft.storage.LogManager;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
import com.alipay.sofa.jraft.util.DisruptorBuilder;
import com.alipay.sofa.jraft.util.DisruptorMetricSet;
import com.alipay.sofa.jraft.util.LogExceptionHandler;
import com.alipay.sofa.jraft.util.NamedThreadFactory;
import com.alipay.sofa.jraft.util.OnlyForTest;
import com.alipay.sofa.jraft.util.Requires;
import com.alipay.sofa.jraft.util.ThreadPoolsFactory;
import com.alipay.sofa.jraft.util.Utils;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.alipay.sofa.jraft.util.concurrent.EventBus;
import com.alipay.sofa.jraft.util.concurrent.EventBusHandler;
import com.alipay.sofa.jraft.util.concurrent.EventBusOptions;

/**
* The finite state machine caller implementation.
Expand Down Expand Up @@ -119,33 +112,15 @@ private static class ApplyTask {
LeaderChangeContext leaderChangeCtx;
Closure done;
CountDownLatch shutdownLatch;

public void reset() {
this.type = null;
this.committedIndex = 0;
this.term = 0;
this.status = null;
this.leaderChangeCtx = null;
this.done = null;
this.shutdownLatch = null;
}
}

private static class ApplyTaskFactory implements EventFactory<ApplyTask> {

@Override
public ApplyTask newInstance() {
return new ApplyTask();
}
}

private class ApplyTaskHandler implements EventHandler<ApplyTask> {
private class ApplyTaskHandler implements EventBusHandler<ApplyTask> {
boolean firstRun = true;
// max committed index in current batch, reset to -1 every batch
private long maxCommittedIndex = -1;

@Override
public void onEvent(final ApplyTask event, final long sequence, final boolean endOfBatch) throws Exception {
public void onEvent(final ApplyTask event, final boolean endOfBatch) throws Exception {
setFsmThread();
this.maxCommittedIndex = runApplyTask(event, this.maxCommittedIndex, endOfBatch);
}
Expand All @@ -170,8 +145,7 @@ private void setFsmThread() {
private volatile TaskType currTask;
private final AtomicLong applyingIndex;
private volatile RaftException error;
private Disruptor<ApplyTask> disruptor;
private RingBuffer<ApplyTask> taskQueue;
private EventBus<ApplyTask> taskEventBus;
private volatile CountDownLatch shutdownLatch;
private NodeMetrics nodeMetrics;
private final CopyOnWriteArrayList<LastAppliedLogIndexListener> lastAppliedLogIndexListeners = new CopyOnWriteArrayList<>();
Expand All @@ -197,20 +171,12 @@ public boolean init(final FSMCallerOptions opts) {
this.lastAppliedIndex.set(opts.getBootstrapId().getIndex());
notifyLastAppliedIndexUpdated(this.lastAppliedIndex.get());
this.lastAppliedTerm = opts.getBootstrapId().getTerm();
this.disruptor = DisruptorBuilder.<ApplyTask> newInstance() //
.setEventFactory(new ApplyTaskFactory()) //
.setRingBufferSize(opts.getDisruptorBufferSize()) //
.setThreadFactory(new NamedThreadFactory("JRaft-FSMCaller-Disruptor-", true)) //
.setProducerType(ProducerType.MULTI) //
.setWaitStrategy(new BlockingWaitStrategy()) //
.build();
this.disruptor.handleEventsWith(new ApplyTaskHandler());
this.disruptor.setDefaultExceptionHandler(new LogExceptionHandler<Object>(getClass().getSimpleName()));
this.taskQueue = this.disruptor.start();
if (this.nodeMetrics.getMetricRegistry() != null) {
this.nodeMetrics.getMetricRegistry().register("jraft-fsm-caller-disruptor",
new DisruptorMetricSet(this.taskQueue));
}

final EventBusOptions eventBusOpts = new EventBusOptions().setMode(opts.getEventBusMode())
.setName("JRaft-FSMCaller-EventBus").setBufferSize(opts.getDisruptorBufferSize())
.setThreadFactory(new NamedThreadFactory("JRaft-FSMCaller-EventBus-", true));
this.taskEventBus = opts.getEventBusFactory().create(eventBusOpts, new ApplyTaskHandler());

this.error = new RaftException(EnumOutter.ErrorType.ERROR_TYPE_NONE);
LOG.info("Starts FSMCaller successfully.");
return true;
Expand All @@ -223,15 +189,16 @@ public synchronized void shutdown() {
}
LOG.info("Shutting down FSMCaller...");

if (this.taskQueue != null) {
if (this.taskEventBus != null) {
final CountDownLatch latch = new CountDownLatch(1);
this.shutdownLatch = latch;

ThreadPoolsFactory.runInThread(getNode().getGroupId(), () -> this.taskQueue.publishEvent((task, sequence) -> {
task.reset();
ThreadPoolsFactory.runInThread(getNode().getGroupId(), () -> {
final ApplyTask task = new ApplyTask();
task.type = TaskType.SHUTDOWN;
task.shutdownLatch = latch;
}));
this.taskEventBus.publish(task);

Copilot AI Dec 18, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The shutdown event publish call lacks error handling. If taskEventBus.publish() fails, the shutdownLatch will never be counted down, causing join() to hang indefinitely. Consider checking the return value or implementing a timeout mechanism.

Suggested change
this.taskEventBus.publish(task);
try {
final boolean published = this.taskEventBus.publish(task);
if (!published) {
LOG.error("Failed to publish SHUTDOWN task to taskEventBus, counting down latch directly.");
latch.countDown();
}
} catch (final Exception e) {
LOG.error("Exception while publishing SHUTDOWN task to taskEventBus, counting down latch directly.", e);
latch.countDown();
}

Copilot uses AI. Check for mistakes.
});
}
}

Expand All @@ -240,91 +207,90 @@ public void addLastAppliedLogIndexListener(final LastAppliedLogIndexListener lis
this.lastAppliedLogIndexListeners.add(listener);
}

private boolean enqueueTask(final EventTranslator<ApplyTask> tpl) {
private boolean enqueueTask(final ApplyTask task) {
if (this.shutdownLatch != null) {
// Shutting down
LOG.warn("FSMCaller is stopped, can not apply new task.");
return false;
}
this.taskQueue.publishEvent(tpl);
return true;
return this.taskEventBus.publish(task);
}
Comment on lines +210 to 217

@coderabbitai coderabbitai Bot Dec 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Find and examine enqueueTask usage in FSMCallerImpl.java
find . -name "FSMCallerImpl.java" -type f | head -5

Repository: sofastack/sofa-jraft

Length of output: 136

🏁 Script executed:

#!/bin/bash
# Read the FSMCallerImpl.java file and search for enqueueTask calls
wc -l ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

Repository: sofastack/sofa-jraft

Length of output: 140

🏁 Script executed:

#!/bin/bash
# Search for all enqueueTask calls in the file with context
rg -n "enqueueTask" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java -A 5 -B 2

Repository: sofastack/sofa-jraft

Length of output: 2779

🏁 Script executed:

#!/bin/bash
# Search for callers of onCommitted, onSnapshotLoad, onSnapshotSave, onLeaderStart, etc.
# First, let's see the full context around the flush method
sed -n '235,250p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

Repository: sofastack/sofa-jraft

Length of output: 548

🏁 Script executed:

#!/bin/bash
# Check how external code calls these FSMCaller methods
# Search for callers of onCommitted, onSnapshotLoad, onSnapshotSave, etc.
rg -n "\.onCommitted\(|\.onSnapshotLoad\(|\.onSnapshotSave\(|\.onLeaderStart\(" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -A 2 | head -60

Repository: sofastack/sofa-jraft

Length of output: 2696

🏁 Script executed:

#!/bin/bash
# Check callers in NodeImpl and BallotBox
# First check BallotBox.commitAt method
sed -n '135,145p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/BallotBox.java

Repository: sofastack/sofa-jraft

Length of output: 382

🏁 Script executed:

#!/bin/bash
# Search for actual calls to fsmCaller or waiter methods from FSMCaller interface
# in other files to see if return values are checked
rg -n "fsmCaller\.(onCommitted|onSnapshotLoad|onSnapshotSave|onLeaderStart|onLeaderStop|onStartFollowing|onStopFollowing)" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -B 2 -A 2

Repository: sofastack/sofa-jraft

Length of output: 2549

🏁 Script executed:

#!/bin/bash
# Check if the callers in NodeImpl check return values from these methods
# Line 1199 - onStopFollowing
sed -n '1195,1210p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

Repository: sofastack/sofa-jraft

Length of output: 753

🏁 Script executed:

#!/bin/bash
# Check line 2935 - onLeaderStart call in NodeImpl
sed -n '2930,2945p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

Repository: sofastack/sofa-jraft

Length of output: 606

🏁 Script executed:

#!/bin/bash
# Check line 3270 - onLeaderStop call in NodeImpl
sed -n '3265,3275p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

Repository: sofastack/sofa-jraft

Length of output: 414

🏁 Script executed:

#!/bin/bash
# Now let's check how BallotBox calls onCommitted
# to see if it checks the return value
sed -n '125,150p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/BallotBox.java

Repository: sofastack/sofa-jraft

Length of output: 1384

🏁 Script executed:

#!/bin/bash
# Let's look at how the flush method (line 244) handles the return value
# and check what taskEventBus.publish does
sed -n '240,250p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

Repository: sofastack/sofa-jraft

Length of output: 430

🏁 Script executed:

#!/bin/bash
# Let's check the publish method and see what it does
# Also check if there's any exception handling
sed -n '210,220p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

Repository: sofastack/sofa-jraft

Length of output: 430

🏁 Script executed:

#!/bin/bash
# Let's verify the actual implementation by checking where taskEventBus is initialized
# and what publish() returns
rg -n "taskEventBus" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java | head -20

Repository: sofastack/sofa-jraft

Length of output: 480

🏁 Script executed:

#!/bin/bash
# Let's check if there are any other callers that DO check return values
# to compare patterns
rg -n "if.*onCommitted|if.*onSnapshotLoad|if.*onSnapshotSave|if.*onLeaderStart|if.*onLeaderStop|if.*onStartFollowing|if.*onStopFollowing" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -B 1 -A 1

Repository: sofastack/sofa-jraft

Length of output: 428

🏁 Script executed:

#!/bin/bash
# Let's check the exact lines in NodeImpl where return values are not checked
grep -n "fsmCaller\.\(onStopFollowing\|onStartFollowing\|onLeaderStart\|onLeaderStop\)" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

Repository: sofastack/sofa-jraft

Length of output: 412

Verify return value handling in FSMCaller method callers to prevent silent task drops.

The concern is valid. Several callers in NodeImpl and BallotBox invoke FSMCaller methods without checking their boolean return values:

  • NodeImpl.resetLeaderId() (lines 1199, 1204): ignores returns from onStopFollowing() and onStartFollowing()
  • NodeImpl (line 2935): ignores return from onLeaderStart()
  • NodeImpl.onLeaderStop() (line 3270): ignores return from onLeaderStop()
  • BallotBox.commitAt() (line 141): ignores return from waiter.onCommitted()
  • FSMCallerImpl.flush() (line 244): ignores return from enqueueTask(), could deadlock if publish fails

When taskEventBus.publish() returns false (e.g., due to overload), tasks are silently dropped. The flush method is particularly problematic—if publish fails, the latch never signals, potentially causing indefinite blocking. All these callers should handle false returns to avoid silent task drops and state machine inconsistencies.


@Override
public boolean hasAvailableCapacity(int requiredCapacity) {
if (this.shutdownLatch != null) {
return false;
}
return this.taskQueue.hasAvailableCapacity(requiredCapacity);
return this.taskEventBus.hasAvailableCapacity(requiredCapacity);
}

@Override
public boolean onCommitted(final long committedIndex) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.COMMITTED;
task.committedIndex = committedIndex;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.COMMITTED;
task.committedIndex = committedIndex;
return enqueueTask(task);
}

/**
* Flush all events in disruptor.
* Flush all events in event bus.
*/
@OnlyForTest
void flush() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
enqueueTask((task, sequence) -> {
task.type = TaskType.FLUSH;
task.shutdownLatch = latch;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.FLUSH;
task.shutdownLatch = latch;
enqueueTask(task);
latch.await();
}

@Override
public boolean onSnapshotLoad(final LoadSnapshotClosure done) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.SNAPSHOT_LOAD;
task.done = done;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.SNAPSHOT_LOAD;
task.done = done;
return enqueueTask(task);
}

@Override
public boolean onSnapshotSave(final SaveSnapshotClosure done) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.SNAPSHOT_SAVE;
task.done = done;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.SNAPSHOT_SAVE;
task.done = done;
return enqueueTask(task);
}

@Override
public boolean onLeaderStop(final Status status) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.LEADER_STOP;
task.status = new Status(status);
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.LEADER_STOP;
task.status = new Status(status);
return enqueueTask(task);
}

@Override
public boolean onLeaderStart(final long term) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.LEADER_START;
task.term = term;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.LEADER_START;
task.term = term;
return enqueueTask(task);
}

@Override
public boolean onStartFollowing(final LeaderChangeContext ctx) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.START_FOLLOWING;
task.leaderChangeCtx = new LeaderChangeContext(ctx.getLeaderId(), ctx.getTerm(), ctx.getStatus());
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.START_FOLLOWING;
task.leaderChangeCtx = new LeaderChangeContext(ctx.getLeaderId(), ctx.getTerm(), ctx.getStatus());
return enqueueTask(task);
}

@Override
public boolean onStopFollowing(final LeaderChangeContext ctx) {
return enqueueTask((task, sequence) -> {
task.type = TaskType.STOP_FOLLOWING;
task.leaderChangeCtx = new LeaderChangeContext(ctx.getLeaderId(), ctx.getTerm(), ctx.getStatus());
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.STOP_FOLLOWING;
task.leaderChangeCtx = new LeaderChangeContext(ctx.getLeaderId(), ctx.getTerm(), ctx.getStatus());
return enqueueTask(task);
}

/**
Expand Down Expand Up @@ -361,10 +327,10 @@ public boolean onError(final RaftException error) {
return false;
}
final OnErrorClosure c = new OnErrorClosure(error);
return enqueueTask((task, sequence) -> {
task.type = TaskType.ERROR;
task.done = c;
});
final ApplyTask task = new ApplyTask();
task.type = TaskType.ERROR;
task.done = c;
return enqueueTask(task);
}

@Override
Expand All @@ -385,7 +351,6 @@ public NodeImpl getNode() {
public synchronized void join() throws InterruptedException {
if (this.shutdownLatch != null) {
this.shutdownLatch.await();
this.disruptor.shutdown();
if (this.afterShutdown != null) {
this.afterShutdown.run(Status.OK());
this.afterShutdown = null;
Expand All @@ -401,7 +366,6 @@ private long runApplyTask(final ApplyTask task, long maxCommittedIndex, final bo
if (task.committedIndex > maxCommittedIndex) {
maxCommittedIndex = task.committedIndex;
}
task.reset();
} else {
if (maxCommittedIndex >= 0) {
this.currTask = TaskType.COMMITTED;
Expand Down Expand Up @@ -458,7 +422,6 @@ private long runApplyTask(final ApplyTask task, long maxCommittedIndex, final bo
}
} finally {
this.nodeMetrics.recordLatency(task.type.metricName(), Utils.monotonicMs() - startMs);
task.reset();
}
}
try {
Expand Down
Loading
Loading