*: add low-latency watermark propagation - #5826
Conversation
Keep the low-latency mode focused on setting the logpuller advance interval to zero. Remove the additional batched heap update path and its helper after the 10k-region E2E test showed comparable mean and p95 latency without it.
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
release-note
|
Warning Review limit reached
Next review available in: 2 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThe PR adds throughput and low-latency server modes. Low-latency mode accelerates heartbeats, subscriptions, checkpoint updates, and metric refreshes. Coordinator state now tracks reporting rounds and node timestamps. ChangesPerformance mode and reporting flow
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ServerConfig
participant MaintainerManager
participant Maintainer
participant DispatcherManager
ServerConfig->>MaintainerManager: select heartbeat interval
ServerConfig->>DispatcherManager: select heartbeat interval and initial delay
MaintainerManager->>Maintainer: process heartbeat
Maintainer->>Maintainer: detect watermark change
Maintainer->>Maintainer: notify checkpoint calculation
DispatcherManager->>DispatcherManager: run heartbeat task
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 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/XL
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
|
/test all |
![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 (2)
logservice/coordinator/coordinator.go (1)
291-319: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSimplify the affected-changefeed loop.
The
len(affectedGIDs) > 0guard at line 306 is redundant. A range over an empty map performs no iterations. Removing the guard reduces one nesting level.♻️ Proposed simplification
- if len(affectedGIDs) > 0 { - for gid := range affectedGIDs { - if state, ok := c.changefeedStates.m[gid]; ok { - if len(state.nodeStates) == 0 || - len(state.nodesReportedSinceLastUpdate) != len(state.nodeStates) { - continue - } - if c.updateChangefeedMetrics(state, pdPhyTs, false) { - state.metricsUpdatedSinceLastTick = true - } - clear(state.nodesReportedSinceLastUpdate) - } - } - } + for gid := range affectedGIDs { + state, ok := c.changefeedStates.m[gid] + if !ok { + continue + } + if len(state.nodeStates) == 0 || + len(state.nodesReportedSinceLastUpdate) != len(state.nodeStates) { + continue + } + if c.updateChangefeedMetrics(state, pdPhyTs, false) { + state.metricsUpdatedSinceLastTick = true + } + clear(state.nodesReportedSinceLastUpdate) + }🤖 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 `@logservice/coordinator/coordinator.go` around lines 291 - 319, Remove the redundant len(affectedGIDs) > 0 guard and place the existing range over affectedGIDs directly around the changefeed-state processing, preserving all current checks and updates inside the loop.logservice/coordinator/coordinator_test.go (1)
408-409: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueClarify the repeated report from
node-1.Line 409 repeats the exact call on line 408 with the same node, time, and lag. The round set already contains
node-1, so the call changes no state. If the intent is to prove that a repeated report does not complete the round, add a comment and an assertion. Otherwise remove the line.♻️ Proposed cleanup
newResolvedTs := report("node-1", pdTime, 50*time.Millisecond) - report("node-1", pdTime, 50*time.Millisecond) + // A repeated report from the same node must not complete the round. + report("node-1", pdTime, 50*time.Millisecond) + require.Len(t, state.nodesReportedSinceLastUpdate, 1) report("node-2", pdTime.Add(300*time.Millisecond), 180*time.Millisecond)As per path instructions: "Prefer focused deterministic tests".
🤖 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 `@logservice/coordinator/coordinator_test.go` around lines 408 - 409, Remove the redundant second report call in the test around newResolvedTs, unless the test intentionally verifies duplicate-report behavior. If retaining it, add a focused comment explaining the scenario and an assertion proving the repeated node-1 report does not complete or otherwise change the round state.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 `@logservice/coordinator/coordinator.go`:
- Around line 344-378: Update the lag calculation in the changefeed metrics
update flow to always set resolvedTsLagGauge from pdPhyTs minus the physical
timestamp of the global minResolvedTs. Remove the maxNodeLag/hasNodeLag tracking
and the conditional override, while preserving the existing force and metrics
initialization behavior.
---
Nitpick comments:
In `@logservice/coordinator/coordinator_test.go`:
- Around line 408-409: Remove the redundant second report call in the test
around newResolvedTs, unless the test intentionally verifies duplicate-report
behavior. If retaining it, add a focused comment explaining the scenario and an
assertion proving the repeated node-1 report does not complete or otherwise
change the round state.
In `@logservice/coordinator/coordinator.go`:
- Around line 291-319: Remove the redundant len(affectedGIDs) > 0 guard and
place the existing range over affectedGIDs directly around the changefeed-state
processing, preserving all current checks and updates inside the loop.
🪄 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 Plus
Run ID: 945b8b8e-ddf8-4092-a95a-19e229879415
📒 Files selected for processing (10)
downstreamadapter/dispatchermanager/task.godownstreamadapter/dispatchermanager/task_test.gologservice/coordinator/coordinator.gologservice/coordinator/coordinator_test.gologservice/eventstore/event_store.gomaintainer/maintainer.gomaintainer/maintainer_manager.gomaintainer/maintainer_test.gopkg/config/server.gopkg/config/server_config_test.go
|
/test all |
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: 3AceShowHand The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
[LGTM Timeline notifier]Timeline:
|
|
|
||
| func heartbeatInterval() time.Duration { | ||
| if config.GetGlobalServerConfig().IsLowLatencyMode() { | ||
| return lowLatencyHeartbeatInterval |
There was a problem hiding this comment.
Can the HeartbeatInterval and HeartbeatInitialDelay parameters be set by the user?
| serverConfig := config.GetGlobalServerConfig() | ||
| resolvedTsAdvanceInterval := int64(serverConfig.KVClient.AdvanceIntervalInMs) | ||
| if serverConfig.IsLowLatencyMode() { | ||
| resolvedTsAdvanceInterval = 0 |
| @@ -699,28 +701,33 @@ func (m *Maintainer) calCheckpointTs(ctx context.Context) { | |||
| case <-ctx.Done(): | |||
| return | |||
| case <-ticker.C: | |||
There was a problem hiding this comment.
And it looks like ticdc can't advance checkpointTs in no Low Latency Mode
asddongmen
added
the
do-not-merge/hold
3 participants
What problem does this PR solve?
Issue Number: ref #5705
This is PR2, the non-EventService subset split from PR1 #5749 so the low-latency propagation changes can be reviewed independently.
PR2 intentionally contains no
pkg/eventservicecode or EventService metric changes; those changes remain in PR1.What is changed and how it works?
performance-mode = "low-latency"while keeping throughput mode as the default and rejecting unknown values.Check List
Tests
GOFLAGS=-mod=mod make unit_test_pkg PKG='<five owning package scopes>'; 459 tests passed).Questions
Will it cause performance regression or break compatibility?
Throughput mode remains the default with unchanged behavior, while low-latency mode intentionally increases heartbeat frequency. Enable it only after all TiCDC nodes are upgraded, and remove the option before downgrading.
Do you need to update user documentation, design documentation or monitoring documentation?
Yes, the performance mode and its resource trade-off need user documentation, which remains tracked by PR1 and is not duplicated in this split.
Release note
Summary by CodeRabbit