Uh oh!
There was an error while loading. Please reload this page.
[Fix](StreamingJob) Optimize CDC consumption strategy - #60181
Conversation
hello-stephen
commented
Jan 23, 2026
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
JNSimba
commented
Jan 23, 2026
run buildall |
There was a problem hiding this comment.
Pull request overview
This PR optimizes the CDC (Change Data Capture) consumption strategy for streaming jobs by refactoring the data polling mechanism and improving heartbeat handling.
Changes:
- Refactored the SourceReader interface to split
readSplitRecordsinto two methods:prepareAndSubmitSplit(for split preparation) andpollRecords(for data retrieval) - Moved polling logic from individual readers to the PipelineCoordinator, introducing heartbeat-based synchronization to determine when to stop polling
- Reduced the Debezium heartbeat interval from 10 seconds to 3 seconds for faster offset updates and connection issue detection
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| Constants.java | Reduced DEBEZIUM_HEARTBEAT_INTERVAL_MS from 10000ms to 3000ms |
| SourceReader.java | Split readSplitRecords into prepareAndSubmitSplit and pollRecords methods |
| SplitReadResult.java | Removed recordIterator field as polling is now handled at coordinator level |
| JdbcIncrementalSourceReader.java | Implemented new split preparation/polling pattern, removed old pollUntilDataAvailable logic |
| MySqlSourceReader.java | Similar refactoring as JdbcIncrementalSourceReader for MySQL-specific implementation |
| PostgresSourceReader.java | Added heartbeat interval configuration for Postgres CDC |
| PipelineCoordinator.java | Major refactoring: added heartbeat-aware polling loops in fetchRecords and writeRecords, extracted helper methods for cleanup and offset extraction |
| StreamingMultiTblTask.java | Minor log message simplification |
Comments suppressed due to low confidence (1)
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:355
- The closeJobStreamLoad method is not synchronized, which could lead to a race condition with getOrCreateBatchStreamLoad. If closeJobStreamLoad is called while another thread is accessing the batch stream load, the stream load could be removed and closed while being used, potentially causing NullPointerException or data loss. Consider synchronizing this method or using a more fine-grained locking mechanism.
public void closeJobStreamLoad(Long jobId) {
DorisBatchStreamLoad batchStreamLoad = batchStreamLoadMap.remove(jobId);
if (batchStreamLoad != null) {
LOG.info("Close DorisBatchStreamLoad for jobId={}", jobId);
batchStreamLoad.close();
batchStreamLoad = null;
}
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
doris-robot
commented
Jan 23, 2026
TPC-H: Total hot run time: 30679 ms |
doris-robot
commented
Jan 23, 2026
TPC-DS: Total hot run time: 172675 ms |
doris-robot
commented
Jan 23, 2026
ClickBench: Total hot run time: 26.61 s |
hello-stephen
commented
Jan 23, 2026
FE UT Coverage ReportIncrement line coverage |
hello-stephen
commented
Jan 23, 2026
FE Regression Coverage ReportIncrement line coverage |
JNSimba
commented
Jan 23, 2026
run buildall |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated 15 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
doris-robot
commented
Jan 23, 2026
TPC-H: Total hot run time: 31211 ms |
doris-robot
commented
Jan 23, 2026
TPC-DS: Total hot run time: 172749 ms |
doris-robot
commented
Jan 23, 2026
ClickBench: Total hot run time: 27.19 s |
hello-stephen
commented
Jan 23, 2026
FE Regression Coverage ReportIncrement line coverage |
JNSimba
commented
Jan 26, 2026
run buildall |
JNSimba
commented
Jan 26, 2026
run buildall |
doris-robot
commented
Jan 26, 2026
TPC-H: Total hot run time: 31339 ms |
doris-robot
commented
Jan 26, 2026
TPC-DS: Total hot run time: 172273 ms |
doris-robot
commented
Jan 26, 2026
ClickBench: Total hot run time: 26.65 s |
JNSimba
commented
Feb 2, 2026
run cloud_p0 |
hello-stephen
commented
Feb 2, 2026
FE Regression Coverage ReportIncrement line coverage |
1 similar comment
hello-stephen
commented
Feb 2, 2026
FE Regression Coverage ReportIncrement line coverage |
JNSimba
commented
Feb 3, 2026
run cloud_p0 |
hello-stephen
commented
Feb 3, 2026
FE Regression Coverage ReportIncrement line coverage |
JNSimba
commented
Feb 3, 2026
run cloud_p0 |
JNSimba
commented
Feb 3, 2026
run buildall |
hello-stephen
commented
Feb 3, 2026
FE Regression Coverage ReportIncrement line coverage |
JNSimba
commented
Feb 3, 2026
run buildall |
JNSimba
commented
Feb 3, 2026
run buildall |
doris-robot
commented
Feb 3, 2026
TPC-H: Total hot run time: 31939 ms |
doris-robot
commented
Feb 3, 2026
ClickBench: Total hot run time: 28.35 s |
PR approved by at least one committer and no changes requested. |
Uh oh!
There was an error while loading. Please reload this page.
### What problem does this PR solve? Related PR: #58898#59461 This PR primarily optimizes the speed of incremental and snapshot reads. 1. For incremental reads: - Binding the fetch logic to an interval allows fetching data within that interval. - Splitting the fetch and write logic asynchronously. 2. For snapshot reads: - Introducing the `snapshot_split_size` and `snapshot_parallelism` parameters. - `snapshot_split_size`: Adjusts the size of each chunk during the split phase, allowing each split to fetch more data. - `snapshot_parallelism`: The degree of parallelism during the snapshot read phase, i.e., how many chunks can run simultaneously, and how many chunks are scheduled in a single task.
hello-stephen
commented
Feb 3, 2026
FE Regression Coverage ReportIncrement line coverage |
…che#60181 (apache#60463) Cherry-picked from apache#60181 Co-authored-by: wudi <wudi@selectdb.com>
What problem does this PR solve?
Issue Number: close #xxx
Related PR: #58898#59461
This PR primarily optimizes the speed of incremental and snapshot reads.
snapshot_split_sizeandsnapshot_parallelismparameters.snapshot_split_size: Adjusts the size of each chunk during the split phase, allowing each split to fetch more data.snapshot_parallelism: The degree of parallelism during the snapshot read phase, i.e., how many chunks can run simultaneously, and how many chunks are scheduled in a single task.Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)