eventcollector: introduce dispatcher session to separate connection lifecycle management#4991
Conversation
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
release-note
📝 WalkthroughWalkthroughThis PR introduces a new dispatcherSession abstraction that encapsulates dispatcher registration lifecycle and event-service coordination logic previously embedded within dispatcherStat. The session manages connection state (current event service, readiness signals, and remote candidates), handles readiness/reset operations via epoch advancement, routes dispatcher events, and provides lifecycle methods for registration and removal. dispatcherStat delegates all connection-state operations to the session and uses per-epoch state tracking for event validation and heartbeat clamping. ChangesSession Abstraction & Dispatcher Lifecycle Refactoring
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 dispatcher heartbeat mechanism to be epoch-aware and moves the heartbeat emission logic from the DispatcherManager to the EventCollector. It introduces a new dispatcherSession to manage connection states and updates the heartbeat protocol to version 2, which now includes epoch information to prevent stale progress reporting and checkpoint jumps. Feedback focuses on improving the robustness of the binary decoding logic in dispatcher_heartbeat.go by adding necessary length checks to prevent potential panics and optimizing slice initializations for better efficiency.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
downstreamadapter/eventcollector/dispatcher_stat.go (1)
233-280:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftReset replay is still filtered by stale commit-ts state.
filterAndUpdateEventByCommitTsstill uses dispatcher-widelastEventCommitTs/gotDDLOnTs/gotSyncpointOnTS, but resets now only replacecurrentEpoch. IfresetTs := target.GetCheckpointTs()lags the last forwarded commit ts, the replay range from the new epoch is<= lastEventCommitTsfrom the old epoch and gets discarded here, so the reset cannot actually backfill the gap it was meant to repair. This dedupe state needs to move into the epoch state, or be reinitialized whenever the epoch advances.🤖 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/eventcollector/dispatcher_stat.go` around lines 233 - 280, filterAndUpdateEventByCommitTs currently uses dispatcher-wide dedupe state (d.lastEventCommitTs, d.gotDDLOnTs, d.gotSyncpointOnTS) which prevents a reset that advances epoch from replaying older commit-ts events; move the dedupe state into the epoch so replayed events aren't filtered by previous epoch values: add lastEventCommitTs/gotDDLOnTs/gotSyncpointOnTS fields to dispatcherEpochState (or ensure those dispatcher-wide flags are reinitialized whenever the epoch in state advances), then update filterAndUpdateEventByCommitTs to read/update the epoch-local fields (e.g., state.lastEventCommitTs, state.gotDDLOnTs, state.gotSyncpointOnTS) instead of d.lastEventCommitTs/d.gotDDLOnTs/d.gotSyncpointOnTS so a reset/backfill can emit events <= previous commit-ts for the new epoch.
🤖 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/eventcollector/dispatcher_session.go`:
- Around line 88-97: The handler advances to the next remote for every non-local
TypeNotReusableEvent even if the signal came from an old/stale eventService, and
getNextRemoteCandidate leaves eventServiceID unchanged when candidates are
exhausted causing setRemoteCandidates to be rejected and the session to stall;
fix by making the non-local not-reusable path first verify the signal's source
matches the current d.eventServiceID (do not advance if it’s stale), and update
getNextRemoteCandidate (and the analogous logic around lines 262-269) to clear
d.eventServiceID (set to empty) when remoteCandidates is empty so subsequent
setRemoteCandidates calls are accepted. Ensure you reference
dispatcherConnState.getNextRemoteCandidate, the d.eventServiceID field and the
non-local TypeNotReusableEvent handling code paths when applying the changes.
- Around line 205-223: The TypeReadyEvent case dereferences event.From at
multiple places without ensuring it's non-nil, causing a panic for malformed
ready signals; update the ready-event branch in dispatcher_session.go (the case
handling commonEvent.TypeReadyEvent, referencing s.isCurrentEventService,
s.removeFrom, localServerID, and s.readyCallback) to first check if event.From
== nil and if so ignore/return, then proceed with the existing logic that
dereferences *event.From; also ensure any existing early checks that compare
event.From to localServerID or call isCurrentEventService use the guarded,
non-nil pointer.
In `@pkg/common/event/dispatcher_heartbeat.go`:
- Around line 48-52: The Unmarshal implementations must validate buffer lengths
before calling buf.Next(...) to avoid panics on short payloads; update
DispatcherProgressLegacy.Unmarshal (and the equivalent legacy and v2 progress
record Unmarshal functions referenced in the review) to check buf.Len() or the
return size before each field extraction (e.g., ensure enough bytes for
DispatcherID.GetSize() and for the 8-byte checkpoint) and return a clear error
on truncated data instead of proceeding to binary.BigEndian.Uint64 or calling
Unmarshal on incomplete slices. Ensure each per-entry length check is added
where DispatcherID.Unmarshal and binary.BigEndian.Uint64 are invoked so
malformed or truncated heartbeats return an error rather than panic.
In `@pkg/eventservice/event_broker.go`:
- Around line 1247-1285: The v1 (DispatcherProgressesLegacy) path doesn't verify
heartbeat.serverID against the dispatcher's current owner, allowing stale
collectors to refresh checkpointTs and lastReceivedHeartbeatTime; update
handleProgress (or its callers) to also validate the dispatcher's owner server
ID: pass heartbeat.serverID into handleProgress (or access it inside) and after
loading dispatcher (dispatcher := dispatcherPtr.Load()) compare the dispatcher's
owner field (e.g. dispatcher.serverID or dispatcher.ownerServerID — the field
that stores the dispatcher's current owner) to heartbeat.serverID and if they
differ either ignore the heartbeat (return) or append a DSStateRemoved response
exactly as done when dispatcherPtr == nil, then proceed with epoch/checkpoint
updates only when the owner matches; apply this change for both
DispatcherProgresses and DispatcherProgressesLegacy paths.
---
Outside diff comments:
In `@downstreamadapter/eventcollector/dispatcher_stat.go`:
- Around line 233-280: filterAndUpdateEventByCommitTs currently uses
dispatcher-wide dedupe state (d.lastEventCommitTs, d.gotDDLOnTs,
d.gotSyncpointOnTS) which prevents a reset that advances epoch from replaying
older commit-ts events; move the dedupe state into the epoch so replayed events
aren't filtered by previous epoch values: add
lastEventCommitTs/gotDDLOnTs/gotSyncpointOnTS fields to dispatcherEpochState (or
ensure those dispatcher-wide flags are reinitialized whenever the epoch in state
advances), then update filterAndUpdateEventByCommitTs to read/update the
epoch-local fields (e.g., state.lastEventCommitTs, state.gotDDLOnTs,
state.gotSyncpointOnTS) instead of
d.lastEventCommitTs/d.gotDDLOnTs/d.gotSyncpointOnTS so a reset/backfill can emit
events <= previous commit-ts for the new epoch.
🪄 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: 02af9288-8f0d-4a4e-9764-f611a48a36dd
📒 Files selected for processing (12)
.gitignoredownstreamadapter/dispatchermanager/dispatcher_manager.godownstreamadapter/eventcollector/dispatcher_session.godownstreamadapter/eventcollector/dispatcher_stat.godownstreamadapter/eventcollector/dispatcher_stat_test.godownstreamadapter/eventcollector/event_collector.godownstreamadapter/eventcollector/event_collector_test.gologservice/eventstore/event_store_test.gopkg/common/event/dispatcher_heartbeat.gopkg/common/event/dispatcher_heartbeat_test.gopkg/eventservice/event_broker.gopkg/eventservice/event_broker_test.go
💤 Files with no reviewable changes (1)
- downstreamadapter/dispatchermanager/dispatcher_manager.go
lidezhu
changed the title
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
downstreamadapter/eventcollector/event_collector.go (1)
424-437:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
groupHeartbeatis missing the empty-ID guard present innewCongestionControlMessages.
newCongestionControlMessages(Line 712–715) explicitly skips dispatchers whosegetEventServiceID()returns"".groupHeartbeatdoes not. BecauseisReceivingDataEvent()andgetEventServiceID()are two separate reads, a concurrentRemoveDispatchercall (which callsstat.remove()beforedispatcherMap.Delete) can race between them, leavinggetEventServiceID()returning"". Thegrouphelper would then insert a heartbeat undernode.ID("")andsendDispatcherHeartbeatwould enqueue a message to an empty target.🛡️ Proposed fix
c.dispatcherMap.Range(func(_, value interface{}) bool { stat := value.(*dispatcherStat) if !stat.isReceivingDataEvent() { return true } + eventServiceID := stat.getEventServiceID() + if eventServiceID == "" { + return true + } checkpointTs, epoch := stat.getHeartbeatProgressForEventService() group( - stat.getEventServiceID(), + eventServiceID, stat.getDispatcherID(), checkpointTs, epoch, ) return true })🤖 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/eventcollector/event_collector.go` around lines 424 - 437, groupHeartbeat can insert heartbeats under an empty node ID because it doesn't guard against stat.getEventServiceID() returning "" (unlike newCongestionControlMessages); update groupHeartbeat's dispatcherMap.Range loop to fetch id := stat.getEventServiceID() after confirming stat.isReceivingDataEvent(), and skip the dispatcher (do not call group or sendDispatcherHeartbeat) if id == "", mirroring the empty-ID check used in newCongestionControlMessages to avoid enqueuing heartbeats for removed/cleared dispatchers (refer to dispatcherMap, dispatcherStat.remove, group, sendDispatcherHeartbeat, getEventServiceID, and isReceivingDataEvent).downstreamadapter/eventcollector/dispatcher_stat_test.go (1)
341-494:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
TestFilterAndUpdateEventByCommitTs: the newly addedexpectedDDLOnTs/expectedSyncOnTs/expectedCommitTsstruct fields are never referenced in the assertions.Lines 489–491 assert the post-call state against the initial input values (
tt.gotDDLOnTs,tt.gotSyncpointOnTS,tt.lastEventCommitTs) rather than the intended expected values. For example, the "DDL event with same commit ts and not got DDL" case hasgotDDLOnTs: falseandexpectedDDLOnTs: true, but the assertion usestt.gotDDLOnTs(i.e.,false). Similarly for "DML event with larger commit ts" whereexpectedCommitTs: 110≠lastEventCommitTs: 100.This means the test does not actually verify state mutations — the
expected*fields are dead code.The fix depends on whether
shouldForwardEventByCommitTsis a pure filter (in which case theexpected*fields should be removed) or also mutates state (in which case the assertions should use the expected fields):🐛 Proposed fix — wire up the expected fields in assertions
- require.Equal(t, tt.gotDDLOnTs, stat.gotDDLOnTs.Load()) - require.Equal(t, tt.gotSyncpointOnTS, stat.gotSyncpointOnTS.Load()) - require.Equal(t, tt.lastEventCommitTs, stat.lastEventCommitTs.Load()) + require.Equal(t, tt.expectedDDLOnTs, stat.gotDDLOnTs.Load()) + require.Equal(t, tt.expectedSyncOnTs, stat.gotSyncpointOnTS.Load()) + require.Equal(t, tt.expectedCommitTs, stat.lastEventCommitTs.Load())Alternatively, if
shouldForwardEventByCommitTsis intentionally a pure filter (withupdateCommitTsStateByEventshandling mutations separately), remove theexpectedDDLOnTs,expectedSyncOnTs,expectedCommitTsfields entirely to avoid misleading readers.🤖 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/eventcollector/dispatcher_stat_test.go` around lines 341 - 494, The test TestFilterAndUpdateEventByCommitTs defines expectedDDLOnTs/expectedSyncOnTs/expectedCommitTs but the assertions check the original input fields (tt.gotDDLOnTs, tt.gotSyncpointOnTS, tt.lastEventCommitTs) so state mutations are never verified; either (A) update the assertions after calling stat.shouldForwardEventByCommitTs to compare against tt.expectedDDLOnTs, tt.expectedSyncOnTs and tt.expectedCommitTs (and keep the existing require.Equal on result), or (B) if shouldForwardEventByCommitTs is meant to be pure, remove the unused expected* fields from the test cases to avoid dead code; reference the test name TestFilterAndUpdateEventByCommitTs and methods shouldForwardEventByCommitTs / updateCommitTsStateByEvents to decide which approach to apply.
🤖 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.
Outside diff comments:
In `@downstreamadapter/eventcollector/dispatcher_stat_test.go`:
- Around line 341-494: The test TestFilterAndUpdateEventByCommitTs defines
expectedDDLOnTs/expectedSyncOnTs/expectedCommitTs but the assertions check the
original input fields (tt.gotDDLOnTs, tt.gotSyncpointOnTS, tt.lastEventCommitTs)
so state mutations are never verified; either (A) update the assertions after
calling stat.shouldForwardEventByCommitTs to compare against tt.expectedDDLOnTs,
tt.expectedSyncOnTs and tt.expectedCommitTs (and keep the existing require.Equal
on result), or (B) if shouldForwardEventByCommitTs is meant to be pure, remove
the unused expected* fields from the test cases to avoid dead code; reference
the test name TestFilterAndUpdateEventByCommitTs and methods
shouldForwardEventByCommitTs / updateCommitTsStateByEvents to decide which
approach to apply.
In `@downstreamadapter/eventcollector/event_collector.go`:
- Around line 424-437: groupHeartbeat can insert heartbeats under an empty node
ID because it doesn't guard against stat.getEventServiceID() returning ""
(unlike newCongestionControlMessages); update groupHeartbeat's
dispatcherMap.Range loop to fetch id := stat.getEventServiceID() after
confirming stat.isReceivingDataEvent(), and skip the dispatcher (do not call
group or sendDispatcherHeartbeat) if id == "", mirroring the empty-ID check used
in newCongestionControlMessages to avoid enqueuing heartbeats for
removed/cleared dispatchers (refer to dispatcherMap, dispatcherStat.remove,
group, sendDispatcherHeartbeat, getEventServiceID, and isReceivingDataEvent).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: fe594fd9-7d83-4488-98f4-781bd0696834
📒 Files selected for processing (5)
downstreamadapter/eventcollector/dispatcher_stat.godownstreamadapter/eventcollector/dispatcher_stat_test.godownstreamadapter/eventcollector/event_collector.godownstreamadapter/eventcollector/event_collector_test.gopkg/sink/mysql/mysql_writer_dml_test.go
✅ Files skipped from review due to trivial changes (1)
- pkg/sink/mysql/mysql_writer_dml_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- downstreamadapter/eventcollector/dispatcher_stat.go
lidezhu
changed the title
There was a problem hiding this comment.
🧹 Nitpick comments (1)
downstreamadapter/eventcollector/dispatcher_stat_test.go (1)
1435-1480: 💤 Low valueLGTM: heartbeat checkpoint progression and clamping to dispatcher checkpointTs are well covered.
The test exercises a clean progression of
maxEventTs: constructor (startTs=100) →doReset(150)→ handshake (180) → resolved (200) → DML (210), and confirms the heartbeat checkpoint clamps tomockDisp.checkPointTswhen it falls belowmaxEventTs(line 1461-1462). InitialmaxEventTs=100correctly reflectsnewDispatcherStat's use oftarget.GetStartTs()per thedispatcher_stat.goconstructor.Minor optional nit: consider adding a sub-assertion for the
currentEventServicevalue alongside the checkpoint comparisons (e.g., afterdoReset) to make the "for event service" intent in the test name explicit, but this is non-blocking.🤖 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/eventcollector/dispatcher_stat_test.go` around lines 1435 - 1480, Add an explicit assertion that the epoch state's currentEventService is set to the event service passed to doReset to make the test intent explicit: after calling stat.doReset(node.ID("event-service-1"), 150) assert that stat.loadCurrentEpochState().currentEventService (or the exported accessor if present) equals node.ID("event-service-1"); this should sit alongside the existing checkpoint assertions that use getHeartbeatCheckpoint() and verifies doReset correctly records the currentEventService for the event service path tested.
🤖 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.
Nitpick comments:
In `@downstreamadapter/eventcollector/dispatcher_stat_test.go`:
- Around line 1435-1480: Add an explicit assertion that the epoch state's
currentEventService is set to the event service passed to doReset to make the
test intent explicit: after calling stat.doReset(node.ID("event-service-1"),
150) assert that stat.loadCurrentEpochState().currentEventService (or the
exported accessor if present) equals node.ID("event-service-1"); this should sit
alongside the existing checkpoint assertions that use getHeartbeatCheckpoint()
and verifies doReset correctly records the currentEventService for the event
service path tested.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 15a68569-5f4b-43f1-a9c1-6c95cf5c3f0e
📒 Files selected for processing (1)
downstreamadapter/eventcollector/dispatcher_stat_test.go
lidezhu
changed the title
|
/test all |
|
Just do some code movement, puller-error-log-view failed because of original logs, it will be handled in later prs. |
ti-chi-bot
Bot
added
needs-1-more-lgtm
|
/gemini summary |
Summary of ChangesThis pull request refactors the dispatcher's connection and lifecycle management by introducing a dedicated Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Activity
|
|
/test all |
|
This PR only involves some code rearrangement with no logic changes. The panic log already exists and will be addressed in a later PR. |
ti-chi-bot
Bot
added
lgtm
and removed
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: flowbehappy, hongyunyan 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 |
[LGTM Timeline notifier]Timeline:
|
|
/retest |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
downstreamadapter/eventcollector/dispatcher_stat_test.go (1)
1371-1398:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
TestHandleBatchDataEventsDoesNotAdvanceCommitTsWhenNoValidEventslikely tests the wrong guard.The test's intent (commit ts not advanced for stale events) requires the event to pass source-ID and readiness checks first. The stat is created without
stat.session.connState.setEventServiceID(...)orreadyEventReceived.Store(true), but the event carriesFrom: createNodeID("service1"). SincehandleBatchDataEventsfilters per-event source (as every other batch-event test demonstrates),"service1" != ""causes the event to be silently dropped at the source check — not at the stale-commit-ts check. The result andrequire.Emptyassertions both pass, but the intended invariant is never exercised.🛠️ Suggested fix — add session setup to test the intended path
stat := newDispatcherStat(mockDisp, nil, nil) stat.lastEventCommitTs.Store(50) stat.currentEpoch.Store(newDispatcherEpochState(10, 1, stat.target.GetStartTs())) +stat.session.connState.setEventServiceID("service1") +stat.session.connState.readyEventReceived.Store(true)🤖 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/eventcollector/dispatcher_stat_test.go` around lines 1371 - 1398, The test TestHandleBatchDataEventsDoesNotAdvanceCommitTsWhenNoValidEvents is currently dropping the event at the source-ID/readiness guard instead of exercising the stale-commit-ts path; to fix it, arrange the session so the event passes source and readiness checks before calling stat.handleBatchDataEvents: set the session event service id to match the event’s source (call stat.session.connState.setEventServiceID with the same createNodeID("service1") used in the event) and mark the session ready (call stat.readyEventReceived.Store(true)); then run the existing assertions to verify handleBatchDataEvents does not advance lastEventCommitTs for the stale commitTs case.
🧹 Nitpick comments (1)
downstreamadapter/eventcollector/dispatcher_stat_test.go (1)
1157-1165: 💤 Low valuePrefer a struct field over string comparison for panic-case branching.
Branching on
tt.name == "multiple events"is fragile — any rename silently converts aPanicsassertion into the normal path. The existingTestHandleSignalEventpattern (which usesexpectedPanic bool) is the right model here.♻️ Proposed refactor
tests := []struct { name string events []dispatcher.DispatcherEvent currentService node.ID lastSeq uint64 lastCommitTs uint64 epoch uint64 want bool + expectPanic bool }{ { - name: "multiple events", + name: "multiple events", + expectPanic: true, events: []dispatcher.DispatcherEvent{ ... }, ... }, ... } ... -if tt.name == "multiple events" { +if tt.expectPanic { require.Panics(t, func() { stat.handleSingleDataEvents(tt.events) }) } else { got := stat.handleSingleDataEvents(tt.events) require.Equal(t, tt.want, got) }🤖 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/eventcollector/dispatcher_stat_test.go` around lines 1157 - 1165, The test currently branches on tt.name == "multiple events" to decide whether to expect a panic when calling stat.handleSingleDataEvents, which is fragile; add a boolean field (e.g. expectedPanic) to the test case struct and use that to choose between require.Panics and normal assertion instead of string-matching tt.name, update all affected cases (including the "multiple events" case) to set expectedPanic=true, and refactor the test body to follow the TestHandleSignalEvent pattern so the panic expectation is explicit and robust.
🤖 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.
Outside diff comments:
In `@downstreamadapter/eventcollector/dispatcher_stat_test.go`:
- Around line 1371-1398: The test
TestHandleBatchDataEventsDoesNotAdvanceCommitTsWhenNoValidEvents is currently
dropping the event at the source-ID/readiness guard instead of exercising the
stale-commit-ts path; to fix it, arrange the session so the event passes source
and readiness checks before calling stat.handleBatchDataEvents: set the session
event service id to match the event’s source (call
stat.session.connState.setEventServiceID with the same createNodeID("service1")
used in the event) and mark the session ready (call
stat.readyEventReceived.Store(true)); then run the existing assertions to verify
handleBatchDataEvents does not advance lastEventCommitTs for the stale commitTs
case.
---
Nitpick comments:
In `@downstreamadapter/eventcollector/dispatcher_stat_test.go`:
- Around line 1157-1165: The test currently branches on tt.name == "multiple
events" to decide whether to expect a panic when calling
stat.handleSingleDataEvents, which is fragile; add a boolean field (e.g.
expectedPanic) to the test case struct and use that to choose between
require.Panics and normal assertion instead of string-matching tt.name, update
all affected cases (including the "multiple events" case) to set
expectedPanic=true, and refactor the test body to follow the
TestHandleSignalEvent pattern so the panic expectation is explicit and robust.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1b3507f3-cd72-4260-8a67-8414e91d4191
📒 Files selected for processing (3)
downstreamadapter/eventcollector/dispatcher_stat.godownstreamadapter/eventcollector/dispatcher_stat_test.godownstreamadapter/eventcollector/event_collector_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- downstreamadapter/eventcollector/event_collector_test.go
- downstreamadapter/eventcollector/dispatcher_stat.go
|
/retest |
3 participants
What problem does this PR solve?
Issue Number: close #4999
What is changed and how it works?
This pull request refactors the dispatcher's connection and lifecycle management by introducing a dedicated
dispatcherSessionabstraction. By moving connection state and coordination logic out ofdispatcherStat, the system achieves better separation of concerns and more robust handling of dispatcher registration, reset operations, and heartbeat synchronization. The changes also improve event validation through explicit per-epoch state tracking, ensuring that dispatchers correctly handle progress monitoring and connection lifecycle events.Highlights
dispatcherSessionto encapsulate dispatcher registration and lifecycle management, decoupling this logic fromdispatcherStat.Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note
Summary by CodeRabbit
Refactor
New Features
Tests