[ISSUE #10898] Unblock the NettyEventExecutor event queue on shutdown - #11005
[ISSUE #10898] Unblock the NettyEventExecutor event queue on shutdown#11005wang-jiahua wants to merge 1 commit into
Conversation
…utdown NettyEventExecutor blocks in eventQueue.poll(3000ms), while ServiceThread signals a stop through the stopped flag plus wakeup(). wakeup() only unparks a thread waiting inside waitForRunning and cannot release a thread blocked on a LinkedBlockingQueue, so the executor keeps waiting until the poll expires before it notices the stopped flag. Every remoting instance therefore spends up to 3 seconds doing nothing on shutdown, and a broker owns several of them. Override wakeup() to offer a sentinel event after delegating to super, so the pending poll returns at once and the loop re-checks the stopped flag. The dispatch loop compares the sentinel by reference and skips it, so no listener callback is triggered and NettyEventType needs no new constant. putNettyEvent is left untouched so genuine channel events keep the queue-size guard. Locally the join time for each NettyEventExecutor drops from about 3000ms to 0ms, BrokerShutdownTest goes from 71.98s to 53.63s, and NettyRemotingAbstractTest from 3.76s to 1.08s. The new test starts the executor, lets it reach the blocking poll, then asserts that shutdown returns in under a second; it fails before this change.
There was a problem hiding this comment.
🟡 Changes recommended
The new test risks flakiness and resource leakage (fixed sleep + missing client cleanup), and wakeup() should avoid enqueuing sentinel events unnecessarily.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR addresses remoting shutdown latency by ensuring NettyRemotingAbstract.NettyEventExecutor can be released immediately from its blocking LinkedBlockingQueue.poll(...) during shutdown, instead of waiting up to the poll timeout.
Changes:
- Add a sentinel
NettyEventand overrideNettyEventExecutor.wakeup()to enqueue it so the blocking poll returns promptly on shutdown. - Update the dispatch loop to skip the sentinel event by reference.
- Add a unit test asserting shutdown completes quickly (under 1s).
File summaries
| File | Description |
|---|---|
| remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java | Enqueues a sentinel in wakeup() and skips it in the event-dispatch loop to unblock shutdown promptly. |
| remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java | Adds a test to validate that NettyEventExecutor.shutdown() does not wait for the queue poll timeout. |
Review details
- Files reviewed: 2/2 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| // done by super.wakeup() cannot release it and shutdown would wait for the poll timeout | ||
| // to expire. Offer a sentinel to make the poll return at once, letting the loop observe | ||
| // the stopped flag immediately. | ||
| this.eventQueue.offer(this.wakeupEvent); |
| NettyRemotingAbstract remoting = new NettyRemotingClient(new NettyClientConfig()); | ||
| remoting.nettyEventExecutor.start(); | ||
|
|
||
| // let the thread reach the blocking poll | ||
| TimeUnit.MILLISECONDS.sleep(300); | ||
|
|
||
| long begin = System.currentTimeMillis(); | ||
| remoting.nettyEventExecutor.shutdown(); | ||
| long elapsed = System.currentTimeMillis() - begin; | ||
|
|
||
| assertThat(remoting.nettyEventExecutor.isStopped()).isTrue(); | ||
| assertThat(elapsed).isLessThan(1000); | ||
| } |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #11005 +/- ##
=============================================
- Coverage 48.80% 48.72% -0.08%
+ Complexity 13764 13743 -21
=============================================
Files 1381 1381
Lines 101574 101578 +4
Branches 13213 13213
=============================================
- Hits 49572 49496 -76
- Misses 45975 46038 +63
- Partials 6027 6044 +17 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Clean fix for the 3-second shutdown delay in NettyEventExecutor. The sentinel-event approach to unblock eventQueue.poll() is a well-known pattern, correctly implemented here: the wakeupEvent is final (safe publication), offer() is thread-safe on LinkedBlockingQueue, and the dispatch loop properly skips the sentinel via reference inequality. super.wakeup() is called first to preserve the existing LockSupport.unpark path. The deterministic test and A/B benchmarks (2700ms → 0.2ms) clearly validate the fix.
Fixes #10898. LGTM.
Automated review by github-manager-bot
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Fixes a 3-second shutdown delay in caused by blocking the thread while cannot release it via .
The sentinel pattern is clean:
- override offers the sentinel to the queue, making return immediately
- Dispatch loop skips the sentinel via identity check ()
- No external subclassing surface (package-private inner class)
Benchmark: shutdown drops from ~2700ms to <1ms. Deterministic test verifies <1s shutdown.
LGTM 👍
Automated review by github-manager-bot
Which Issue(s) This PR Fixes
Fixes #10898
Brief Description
NettyEventExecutorblocks ineventQueue.poll(3000ms)instead ofwaitForRunning, so theLockSupport.unparkperformed byServiceThread.shutdown()cannot release it: the thread only notices the stopped flag once the poll expires, and every shutdown of a remoting instance pays up to 3 seconds for nothing.This PR overrides
wakeup()to offer a sentinelNettyEventinto the queue aftersuper.wakeup(), making the blocking poll return immediately; the dispatch loop skips the sentinel. No behavior change apart from the faster exit.How Did You Test This Change?
testNettyEventExecutorShutdownDoesNotWaitForPollTimeout: on current develop it fails after 3.66s (the thread waits out the full poll timeout); with this patch it passes in under 1s.NettyRemotingAbstractTest6/6, checkstyle clean.NettyEventExecutorstart/shutdown directly, 5 runs per arm:NettyEventExecutoris a package-private inner class, so there is no external subclassing surface affected by the newwakeup()override.