*: add low-latency performance mode#5749
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.
Replace the periodic schema-capped scan retry with applied SchemaStore notifications. Serialize dispatcher scan scheduling with a short-lock state machine and coalesce worker continuations without dropping queued work.
Keep no-DML/no-DDL resolved notifications out of the scan worker queue while preserving dispatcher scan ownership. Gate continuation and schema-blocked recovery on low-latency mode, and cover queue-full recovery with a dropped-task metric.
ti-chi-bot
Bot
added
the
release-note
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: 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 |
ti-chi-bot
Bot
added
the
size/XXL
📝 WalkthroughWalkthroughThe PR introduces throughput and low-latency server modes, configurable heartbeat and manager timing, event-driven checkpoint propagation, reporting-round-aware coordinator metrics, and an event-service scan state machine with schema-blocked retries and queue recovery. ChangesLow-latency performance mode
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant DispatcherManager
participant Maintainer
participant EventBroker
participant SchemaStore
DispatcherManager->>Maintainer: send heartbeat watermark
Maintainer->>Maintainer: signal checkpoint recalculation
Maintainer->>EventBroker: request dispatcher scan
EventBroker->>SchemaStore: check DDL frontier
SchemaStore-->>EventBroker: return scan range or blocked frontier
EventBroker->>EventBroker: enqueue or retry scan
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@maintainer/maintainer.go`:
- Around line 961-964: Update the status-handling flow around HandleStatus to
always call notifyCheckpointUpdate after applying req.Statuses, rather than only
when watermarkUpdated is true. Preserve the existing HandleStatus call and rely
on the buffered notification channel to coalesce duplicate wakeups.
In `@pkg/eventservice/event_broker.go`:
- Around line 1056-1062: Update the blocking enqueue in prepareScanFromNotify to
select between sending d to c.taskChan[d.scanWorkerIndex] and broker shutdown
via the broker-level done channel or errgroup context. If shutdown wins, reset
the dispatcher’s queued scan state, including the associated busy/queued
markers, before returning so it cannot remain permanently queued.
🪄 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: 998e1e16-fee4-4cdc-b75c-0a3c2459d3cc
📒 Files selected for processing (15)
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.gopkg/eventservice/dispatcher_stat.gopkg/eventservice/event_broker.gopkg/eventservice/event_broker_test.gopkg/eventservice/metrics_collector.gopkg/metrics/event_service.go
| m.controller.HandleStatus(msg.From, req.Statuses) | ||
| if watermarkUpdated { | ||
| m.notifyCheckpointUpdate() | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Wake checkpoint calculation after status updates.
HandleStatus can change scheduler or barrier constraints used by calculateNewCheckpointTs, but this only signals recalculation when the watermark cache changed. A status-only or duplicate-watermark heartbeat therefore waits for the periodic ticker in low-latency mode. Notify after handling statuses as well; the buffered channel already coalesces redundant wakeups.
🤖 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 `@maintainer/maintainer.go` around lines 961 - 964, Update the status-handling
flow around HandleStatus to always call notifyCheckpointUpdate after applying
req.Statuses, rather than only when watermarkUpdated is true. Preserve the
existing HandleStatus call and rely on the buffered notification channel to
coalesce duplicate wakeups.
| d.scanState = dispatcherScanQueued | ||
| d.schemaBlockedUntilTs = 0 | ||
| d.scanMu.Unlock() | ||
|
|
||
| // Only the external EventStore notify path may wait for capacity. Scan workers | ||
| // use requestScan so that they never block on their own queue. | ||
| c.taskChan[d.scanWorkerIndex] <- d |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Blocking enqueue has no shutdown escape.
prepareScanFromNotify runs on the EventStore notify callback goroutine and blocks unconditionally on c.taskChan[d.scanWorkerIndex] <- d. After cancel() the scan workers return, so any notify in flight (or arriving before dispatchers are unregistered) blocks forever, leaking the notify goroutine and pinning the dispatcher in dispatcherScanQueued (permanently isScanBusy). Select on a broker-level done channel (or the errgroup ctx) alongside the send and bail out by resetting the state.
🛡️ Sketch
- // Only the external EventStore notify path may wait for capacity. Scan workers
- // use requestScan so that they never block on their own queue.
- c.taskChan[d.scanWorkerIndex] <- d
+ // Only the external EventStore notify path may wait for capacity. Scan workers
+ // use requestScan so that they never block on their own queue.
+ select {
+ case c.taskChan[d.scanWorkerIndex] <- d:
+ case <-c.done: // closed by eventBroker.close()
+ d.scanMu.Lock()
+ if d.scanState == dispatcherScanQueued {
+ d.scanState = dispatcherScanIdle
+ }
+ d.scanMu.Unlock()
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| d.scanState = dispatcherScanQueued | |
| d.schemaBlockedUntilTs = 0 | |
| d.scanMu.Unlock() | |
| // Only the external EventStore notify path may wait for capacity. Scan workers | |
| // use requestScan so that they never block on their own queue. | |
| c.taskChan[d.scanWorkerIndex] <- d | |
| d.scanState = dispatcherScanQueued | |
| d.schemaBlockedUntilTs = 0 | |
| d.scanMu.Unlock() | |
| // Only the external EventStore notify path may wait for capacity. Scan workers | |
| // use requestScan so that they never block on their own queue. | |
| select { | |
| case c.taskChan[d.scanWorkerIndex] <- d: | |
| case <-c.done: // closed by eventBroker.close() | |
| d.scanMu.Lock() | |
| if d.scanState == dispatcherScanQueued { | |
| d.scanState = dispatcherScanIdle | |
| } | |
| d.scanMu.Unlock() | |
| } |
🤖 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 `@pkg/eventservice/event_broker.go` around lines 1056 - 1062, Update the
blocking enqueue in prepareScanFromNotify to select between sending d to
c.taskChan[d.scanWorkerIndex] and broker shutdown via the broker-level done
channel or errgroup context. If shutdown wins, reset the dispatcher’s queued
scan state, including the associated busy/queued markers, before returning so it
cannot remain permanently queued.
|
/test all |
|
@asddongmen: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
1 participant
What problem does this PR solve?
Issue Number: close #5705
Timestamp batching adds avoidable latency to changefeeds with many dispatchers
or Regions.
What is changed and how it works?
Add opt-in
performance-mode = "low-latency"for eager LogPuller advancement,watermark reporting, checkpoint calculation, and no-DML/DDL EventBroker
notifications. Throughput remains the default; scan recovery is
low-latency-only.
Three-capture, 10-minute results; values are low-latency/throughput/master.
Resolved is maintainer resolved-ts lag; CPU and RSS are aggregate.
This reduces checkpoint/resolved lag by 64.0%/48.7% with +3.0% CPU and -2.4%
RSS. Table-scale latency gains cost 23-34% CPU; RSS stays within 4%.
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
No compatibility change. Throughput remains default; its CPU/RSS stayed within
2.2% of master.
Do you need to update user documentation, design documentation or monitoring documentation?
Yes, document the mode and resource trade-off.
Release note
Summary by CodeRabbit
New Features
Bug Fixes