downstreamadapter: harden table trigger takeover#5436
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds maintainer-epoch fields to redo protobuf messages, passes sender metadata through redo/control handlers, and updates bootstrap orchestration to validate maintainer updates and trigger state. Tests and one integration script expand coverage for the new epoch-aware paths. ChangesMaintainer Epoch Fencing for Redo Messages and Bootstrap
Sequence Diagram(s)sequenceDiagram
participant Client
participant DispatcherOrchestrator
participant DispatcherManager
Client->>DispatcherOrchestrator: bootstrap request
DispatcherOrchestrator->>DispatcherManager: CanUpdateMaintainer(from, epoch)
DispatcherOrchestrator->>DispatcherManager: validate / ensure table triggers
DispatcherManager-->>DispatcherOrchestrator: accepted or stale/error
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the bootstrap request handling in DispatcherOrchestrator by extracting helper functions to verify and initialize table trigger event and redo dispatchers, and updates operator retrieval to properly handle stale remove operators. It also adds extensive unit tests and updates an integration test to search logs across all CDC nodes. The review feedback highlights a critical mutex leak in handlePostBootstrapRequest due to a missing unlock when fenced, as well as a regression in the new helper functions which fail to gracefully handle write path closed errors during shutdown.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if m.fenced.Load() { | ||
| manager.MaintainerFenceMu.Unlock() | ||
| manager.LocalFence() | ||
| return nil | ||
| } |
There was a problem hiding this comment.
In handlePostBootstrapRequest, if m.fenced.Load() is true, the function returns early without unlocking manager.MaintainerFenceMu. This will cause a mutex leak and potential deadlocks during shutdown or subsequent requests. We should unlock manager.MaintainerFenceMu before calling manager.LocalFence() and returning.
| if m.fenced.Load() { | |
| manager.MaintainerFenceMu.Unlock() | |
| manager.LocalFence() | |
| return nil | |
| } | |
| if m.fenced.Load() { | |
| manager.MaintainerFenceMu.Unlock() | |
| manager.LocalFence() | |
| return nil | |
| } |
| if err := manager.NewTableTriggerEventDispatcher(id, startTs, false); err != nil { | ||
| log.Error("failed to create table trigger event dispatcher", | ||
| zap.Stringer("changefeedID", cfId), zap.Error(err)) | ||
| return err | ||
| } |
There was a problem hiding this comment.
If manager.NewTableTriggerEventDispatcher fails because the write path is closed, it is an expected state during shutdown or fencing. Logging it as an Error and returning the error to the maintainer is a regression from the original behavior (where it was logged as Info and returned nil). We should handle IsWritePathClosedError specially and return nil to allow clean shutdown.
if err := manager.NewTableTriggerEventDispatcher(id, startTs, false); err != nil {
if dispatchermanager.IsWritePathClosedError(err) {
log.Info("dispatcher manager write path closed while creating table trigger event dispatcher",
zap.Stringer("changefeedID", cfId), zap.Error(err))
return nil
}
log.Error("failed to create table trigger event dispatcher",
zap.Stringer("changefeedID", cfId), zap.Error(err))
return err
}| if err := manager.NewTableTriggerRedoDispatcher(id, startTs, false); err != nil { | ||
| log.Error("failed to create table trigger redo dispatcher", | ||
| zap.Stringer("changefeedID", cfId), zap.Error(err)) | ||
| return err | ||
| } |
There was a problem hiding this comment.
Similarly, if manager.NewTableTriggerRedoDispatcher fails because the write path is closed, we should handle IsWritePathClosedError specially and return nil instead of logging an Error and returning the error to the maintainer.
if err := manager.NewTableTriggerRedoDispatcher(id, startTs, false); err != nil {
if dispatchermanager.IsWritePathClosedError(err) {
log.Info("dispatcher manager write path closed while creating table trigger redo dispatcher",
zap.Stringer("changefeedID", cfId), zap.Error(err))
return nil
}
log.Error("failed to create table trigger redo dispatcher",
zap.Stringer("changefeedID", cfId), zap.Error(err))
return err
}…le-maintainer-fence
…2-trigger-takeover
…le-maintainer-fence
…2-trigger-takeover
…le-maintainer-fence
…2-trigger-takeover
…le-maintainer-fence
…2-trigger-takeover
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
hongyunyan
changed the base branch from
codex/pr-5182-stale-maintainer-fence
to
master
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go (1)
461-463: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep trigger logs as constant messages.
triggerNameis internal today, so this is not an injection finding; still, constant messages with structured fields avoid SAST noise and make log queries more stable. As per coding guidelines, “Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs.”Proposed refactor
- log.Error("failed to create "+triggerName, - zap.Stringer("changefeedID", cfId), zap.Error(err)) + log.Error("failed to create table trigger dispatcher", + zap.Stringer("changefeedID", cfId), + zap.String("triggerName", triggerName), + zap.Error(err))- log.Error(triggerName+" id mismatch during bootstrap", + log.Error("table trigger dispatcher id mismatch during bootstrap", zap.Stringer("changefeedID", cfId), + zap.String("triggerName", triggerName), zap.Stringer("expectedDispatcherID", expectedID), zap.Stringer("actualDispatcherID", current.GetId()))- log.Info("dispatcher manager write path closed while creating "+triggerName, - zap.Stringer("changefeedID", cfId), zap.Error(err)) + log.Info("dispatcher manager write path closed while creating table trigger dispatcher", + zap.Stringer("changefeedID", cfId), + zap.String("triggerName", triggerName), + zap.Error(err))Also applies to: 481-484, 497-499
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go` around lines 461 - 463, The trigger creation error logs in DispatcherOrchestrator should use constant messages instead of concatenating triggerName into the log text. Update the affected log statements in the create/stop/cleanup paths to keep a fixed message and move triggerName into structured fields alongside changefeedID and the error, so the DispatcherOrchestrator logs stay stable and SAST-friendly.Sources: Coding guidelines, Linters/SAST tools
downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go (2)
621-623: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winWait for
DispatcherManagershutdown in cleanup.
TryClose(false)is asynchronous here, so these cleanups can return while the manager goroutines are still running against services that the test is already tearing down. Wrap the cleanup inrequire.Eventually(or equivalent) untilTryClose(false)reports closed.Suggested change
t.Cleanup(func() { - manager.TryClose(false) + require.Eventually(t, func() bool { + return manager.TryClose(false) + }, time.Second, 10*time.Millisecond) })As per coding guidelines, "
**/*_test.go: Prefer focused deterministic tests; see docs/agents/testing.md before adding or changing tests."Also applies to: 690-692, 757-759, 817-819, 865-867
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go` around lines 621 - 623, The cleanup in the DispatcherManager tests is returning before asynchronous shutdown completes, which can leave goroutines running against torn-down services. Update each affected t.Cleanup that calls TryClose(false) in dispatcher_orchestrator_test.go to wait until the manager is actually closed, using require.Eventually or an equivalent polling check against the DispatcherManager shutdown state. Keep the change localized to the test helpers/cleanup blocks that reference manager.TryClose(false).Source: Coding guidelines
775-776: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winBound mock message reads with a timeout.
These raw channel receives will hang the test binary when the response is missing. A small
requireMessagehelper withselect/time.Afterwould make failures deterministic instead of surfacing as package timeouts.As per coding guidelines, "
**/*_test.go: Prefer focused deterministic tests; see docs/agents/testing.md before adding or changing tests."Also applies to: 788-789, 832-833, 943-945
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go` around lines 775 - 776, The raw receives from mc.GetMessageChannel() in dispatcher_orchestrator_test.go can hang the test when a message is missing. Update the affected test cases to use a small requireMessage helper that waits with select and time.After, then unwraps the message for the existing bootstrapResponseMsg/bootstrapResponse assertions. Apply the same bounded receive pattern to the other listed channel reads in the dispatcher orchestrator tests so failures become deterministic instead of package timeouts.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go`:
- Around line 367-374: Reject nil-trigger bootstraps in DispatcherOrchestrator
when the reused manager still has a different existing trigger. Update the
nil-id handling in dispatcher_orchestrator.go within the dispatcher_orchestrator
logic so that the nil path checks the manager’s current trigger state before
returning success, instead of blindly allowing ownership transfer. Use the
existing bootstrap/maintainer flow around the nil-id branch and the ownership
transfer path that reaches the trigger handoff logic to ensure a reused manager
only proceeds when it has no trigger or already matches the requested trigger
ID.
---
Nitpick comments:
In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go`:
- Around line 621-623: The cleanup in the DispatcherManager tests is returning
before asynchronous shutdown completes, which can leave goroutines running
against torn-down services. Update each affected t.Cleanup that calls
TryClose(false) in dispatcher_orchestrator_test.go to wait until the manager is
actually closed, using require.Eventually or an equivalent polling check against
the DispatcherManager shutdown state. Keep the change localized to the test
helpers/cleanup blocks that reference manager.TryClose(false).
- Around line 775-776: The raw receives from mc.GetMessageChannel() in
dispatcher_orchestrator_test.go can hang the test when a message is missing.
Update the affected test cases to use a small requireMessage helper that waits
with select and time.After, then unwraps the message for the existing
bootstrapResponseMsg/bootstrapResponse assertions. Apply the same bounded
receive pattern to the other listed channel reads in the dispatcher orchestrator
tests so failures become deterministic instead of package timeouts.
In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go`:
- Around line 461-463: The trigger creation error logs in DispatcherOrchestrator
should use constant messages instead of concatenating triggerName into the log
text. Update the affected log statements in the create/stop/cleanup paths to
keep a fixed message and move triggerName into structured fields alongside
changefeedID and the error, so the DispatcherOrchestrator logs stay stable and
SAST-friendly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7d6c52a7-c803-41bd-8f9e-3650e5277c16
⛔ Files ignored due to path filters (1)
heartbeatpb/heartbeat.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (9)
downstreamadapter/dispatchermanager/dispatcher_manager_info.godownstreamadapter/dispatchermanager/heartbeat_collector.godownstreamadapter/dispatchermanager/helper.godownstreamadapter/dispatchermanager/helper_test.godownstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.godownstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.goheartbeatpb/heartbeat.protomaintainer/maintainer.gotests/integration_tests/in_flight_syncpoint_during_scheduling/run.sh
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go`:
- Around line 550-557: The post-bootstrap path in DispatcherOrchestrator is
returning nil before verifying whether the request is current, which can leave
the current maintainer without the expected error response. Update the
post-bootstrap handling in the relevant dispatcher flow to call
IsMaintainerRequestAllowed first, then if the table trigger dispatcher is
missing route the failure through handleDispatcherError instead of returning
nil. Keep using the repository’s predefined error types when constructing or
propagating the missing-trigger error.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7a7069e0-a17a-4cd6-9ff2-399308f36076
📒 Files selected for processing (2)
downstreamadapter/dispatchermanager/helper.godownstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go
🚧 Files skipped from review as they are similar to previous changes (1)
- downstreamadapter/dispatchermanager/helper.go
ti-chi-bot
Bot
added
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: 3AceShowHand, asddongmen The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
Bot
added
lgtm
and removed
needs-1-more-lgtm
[LGTM Timeline notifier]Timeline:
|
3 participants
What problem does this PR solve?
Issue Number: close #5083
This is PR 3 of 3 split from #5182 and is stacked on PR 2.
Background:
PR 1 persists maintainer epochs before ownership changes. PR 2 fences stale
maintainer control messages at the maintainer and dispatcher-manager boundary.
This final PR hardens the table trigger takeover cases that still depend on
dispatcher-manager bootstrap recovery.
Motivation:
During a same-capture higher-epoch maintainer replacement, a reused dispatcher
manager can still have table trigger event or redo dispatchers from the previous
owner. Replacing a mismatched trigger in place can duplicate DDL ownership, while
dropping stale remove operators during bootstrap can lose the cleanup intent for
an already reported dispatcher.
What is changed and how it works?
This PR adds takeover-specific hardening:
dispatcher yet or already has the trigger dispatcher requested by the current
bootstrap owner.
rejects mismatched trigger IDs instead of replacing them in place.
bootstrap snapshot reports the dispatcher they remove.
close acknowledgements, closed-epoch tombstones, and table trigger ID mismatch
handling.
node logs because the merge task may run on any dispatcher-manager owner after
scheduling.
Stack:
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
No expected performance regression. The added checks run during maintainer
bootstrap and bootstrap response reconstruction, not on the event write path.
The behavior is compatible with rolling upgrades because it builds on the epoch
0 compatibility rules introduced in the previous PRs.
Do you need to update user documentation, design documentation or monitoring documentation?
No.
Release note
Validation
make fmtgo test ./downstreamadapter/dispatcherorchestrator ./downstreamadapter/dispatchermanagerSummary by CodeRabbit
maintainer_epochfor tighter stale protection.