Repository navigation
[feature](load) Track Kinesis shard lifecycle and durable starting positions - #68357
Closed
0AyanamiRei wants to merge 9 commits into
Closed
0AyanamiRei wants to merge 9 commits into
0AyanamiRei wants to merge 9 commits into
Conversation
0AyanamiRei
requested review from
924060929,
CalvinKirs,
gavinchou,
liaoxin01 and
mymeiyi
as code owners
September 22, 2026 02:17
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
0AyanamiRei
added a commit
to 0AyanamiRei/doris
that referenced
this pull request
Sep 22, 2026
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: A confirmed child shard currently waits for all parents to finish and become visible, although this Kinesis ingestion design does not require ordering across parent and child shards. Remove PENDING_PARENT and the parent completion check so shards with confirmed metadata and resolved positions can consume independently. Preserve lineage, initial-snapshot position selection, per-shard EOF persistence at COMMITTED and completion at VISIBLE, and whole-job pause when a confirmed unfinished shard disappears. Update topology and job-level tests for independent split/merge scheduling, task ownership, transaction callbacks, replay, and missing-shard pause behavior. Repair existing Kinesis test fixtures so the complete FE Kinesis suite exercises the topology path and valid restoration dependencies. ### Release note Allow Kinesis child shards to consume while parent shards drain, without waiting for parent transaction visibility. ### Check List (For Author) - Test: Unit Test - FE: MAVEN_ARGS='-Dmaven.compiler.useIncrementalCompilation=false' ./run-fe-ut.sh --run 'Kinesis*Test' (58 passed) - BE: ./run-be-ut.sh --run --filter='*Kinesis*' -j48 (18 passed) - Checkstyle 9.3 with repository rules passed for all modified Java files. - Real AWS and FE/BE end-to-end regression were not run. - Behavior changed: Yes (remove cross-shard parent completion barriers; preserve per-shard transaction lifecycle) - Does this need documentation: Yes (update shard state-machine design to remove PENDING_PARENT)
0AyanamiRei
changed the title
0AyanamiRei
added a commit
to 0AyanamiRei/doris
that referenced
this pull request
Sep 22, 2026
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: Both nonterminal GetRecords response branches advance the same shard iterator. Move the shared increment after the optional empty-page log. Remove the unused KinesisDataConsumer child-parent map, accessor, and reset cleanup. Retain the EOF marker and consumer-group lineage aggregation used by transaction attachments. ### Release note None ### Check List (For Author) - Test: Unit Test - ./run-be-ut.sh --run --filter='*Kinesis*' -j48: 18 passed. - Header hygiene and clang-format 16 checks on changed lines passed. - clang-tidy reports existing diagnostics outside this patch, including an unmatched NOLINTEND in core/types.h. - Behavior changed: No - Does this need documentation: No
0AyanamiRei
added a commit
to 0AyanamiRei/doris
that referenced
this pull request
Sep 22, 2026
### What problem does this PR solve? Issue Number: None Related PR: apache#68357, apache#67869 Problem Summary: When the consumer group reaches a batch limit during enqueueing, the helper previously returned OK without telling group_consume to stop. A multi-shard consumer could issue further AWS requests while the group waited for it to exit; an unnecessary source error could fail an otherwise valid appended prefix. Return an explicit COMPLETE or QUEUE_SHUTDOWN enqueue result and exit the whole consumer successfully on shutdown. Check cancellation and the time budget between shard requests as well as between polling rounds. Add deterministic multi-shard tests through the real consumer/group methods, covering graceful batch completion, prevention of subsequent source failures, queue shutdown before cancellation, and cancellation after an empty page. ### Release note Avoid unnecessary Kinesis requests and batch failures after Routine Load reaches its batch limit. ### Check List (For Author) - Test: Unit Test - Before: ./run-be-ut.sh --run --filter='KinesisConsumerShutdownTest.*' -j48 (all 4 new cases reproduced the bug) - After: ./run-be-ut.sh --run --filter='*Kinesis*' -j48 (22 passed) - Header hygiene and changed-line clang-format 16 checks passed. - New shutdown test file passed clang-tidy; existing diagnostics in unchanged source/header code still prevent a clean whole-file result for data_consumer.cpp. - No live AWS or FE/BE end-to-end regression was run. - Behavior changed: Yes (a stopped enqueue ends the consumer successfully before subsequent shard requests) - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#67678, apache#67696, apache#67869, apache#67874, apache#67879 Problem Summary: Kinesis shards can split and merge while a routine load job is running. Maintaining only open/closed shard lists and progress cannot express parent-child lineage or the COMMITTED-to-VISIBLE completion barrier, which can schedule children too early, lose lineage, or recreate completed parents. This change adds a durable shard topology state machine, concrete initial-position persistence, idempotent journal/image replay, and Kinesis BE progress handling for EOF and ChildShards metadata. New FE, BE, and regression coverage verifies shard transitions, transaction visibility, latest-position resolution, reshard boundaries, and deterministic Kinesis upstream behavior. ### Release note Add parent-child shard topology tracking and visibility-gated scheduling for Kinesis Routine Load. ### Check List (For Author) - Test: Unit Test - `./run-be-ut.sh --run --filter='*Kinesis*' -j48` - Result: 18 Kinesis tests passed - No FE unit test or AWS regression test was run in this environment. - Behavior changed: Yes (Kinesis Routine Load now schedules shards through the topology state machine.) - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: A confirmed child shard currently waits for all parents to finish and become visible, although this Kinesis ingestion design does not require ordering across parent and child shards. Remove PENDING_PARENT and the parent completion check so shards with confirmed metadata and resolved positions can consume independently. Preserve lineage, initial-snapshot position selection, per-shard EOF persistence at COMMITTED and completion at VISIBLE, and whole-job pause when a confirmed unfinished shard disappears. Update topology and job-level tests for independent split/merge scheduling, task ownership, transaction callbacks, replay, and missing-shard pause behavior. Repair existing Kinesis test fixtures so the complete FE Kinesis suite exercises the topology path and valid restoration dependencies. ### Release note Allow Kinesis child shards to consume while parent shards drain, without waiting for parent transaction visibility. ### Check List (For Author) - Test: Unit Test - FE: MAVEN_ARGS='-Dmaven.compiler.useIncrementalCompilation=false' ./run-fe-ut.sh --run 'Kinesis*Test' (58 passed) - BE: ./run-be-ut.sh --run --filter='*Kinesis*' -j48 (18 passed) - Checkstyle 9.3 with repository rules passed for all modified Java files. - Real AWS and FE/BE end-to-end regression were not run. - Behavior changed: Yes (remove cross-shard parent completion barriers; preserve per-shard transaction lifecycle) - Does this need documentation: Yes (update shard state-machine design to remove PENDING_PARENT)
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: Both nonterminal GetRecords response branches advance the same shard iterator. Move the shared increment after the optional empty-page log. Remove the unused KinesisDataConsumer child-parent map, accessor, and reset cleanup. Retain the EOF marker and consumer-group lineage aggregation used by transaction attachments. ### Release note None ### Check List (For Author) - Test: Unit Test - ./run-be-ut.sh --run --filter='*Kinesis*' -j48: 18 passed. - Header hygiene and clang-format 16 checks on changed lines passed. - clang-tidy reports existing diagnostics outside this patch, including an unmatched NOLINTEND in core/types.h. - Behavior changed: No - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#68357, apache#67869 Problem Summary: When the consumer group reaches a batch limit during enqueueing, the helper previously returned OK without telling group_consume to stop. A multi-shard consumer could issue further AWS requests while the group waited for it to exit; an unnecessary source error could fail an otherwise valid appended prefix. Return an explicit COMPLETE or QUEUE_SHUTDOWN enqueue result and exit the whole consumer successfully on shutdown. Check cancellation and the time budget between shard requests as well as between polling rounds. Add deterministic multi-shard tests through the real consumer/group methods, covering graceful batch completion, prevention of subsequent source failures, queue shutdown before cancellation, and cancellation after an empty page. ### Release note Avoid unnecessary Kinesis requests and batch failures after Routine Load reaches its batch limit. ### Check List (For Author) - Test: Unit Test - Before: ./run-be-ut.sh --run --filter='KinesisConsumerShutdownTest.*' -j48 (all 4 new cases reproduced the bug) - After: ./run-be-ut.sh --run --filter='*Kinesis*' -j48 (22 passed) - Header hygiene and changed-line clang-format 16 checks passed. - New shutdown test file passed clang-tidy; existing diagnostics in unchanged source/header code still prevent a clean whole-file result for data_consumer.cpp. - No live AWS or FE/BE end-to-end regression was run. - Behavior changed: Yes (a stopped enqueue ends the consumer successfully before subsequent shard requests) - Does this need documentation: No
0AyanamiRei
force-pushed
the
wip/kinesis-shard-topology-20260916
branch
from
September 23, 2026 02:57
7a9ee1b to
7635434
Compare
Contributor
|
run buildall |
Contributor
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Static review of exact head 7635434adfda681c2d1d8623ee8662c3780b3931. Requesting changes for four P1 issues and one P2 issue.
Checkpoint conclusions:
- Goal, scope, and data correctness: the topology/EOF/LATEST design is cohesive and all 39 changed paths were reviewed, but supported rolling upgrades, legacy image restoration, and source-changing ALTERs have correctness gaps called out inline.
- Concurrency and lifecycle: normal success/error/cancellation/shutdown and COMMITTED/VISIBLE/replay paths were traced; retained-runnable submission can still complete an RPC early, while the remaining transaction/topology lifecycle is coherent.
- Persistence, compatibility, and crash/replay behavior: new journal operations and new-format replay ordering are wired consistently, but protobuf tags are reused incompatibly and pre-topology image state is not migrated.
- Configuration, FE/BE propagation, parallel paths, observability, and performance: new timeout/pool settings propagate to their consumers, errors are surfaced, Cloud mode is explicitly unsupported, and the shared pool is bounded; the scan callbacks nonetheless bypass non-Orphan memory accounting. No additional actionable issue was found in these checkpoints.
- Tests and expected results: the changed tests cover the new topology, LATEST, EOF, cancellation, replay, and restart paths, but do not cover the five failing boundaries identified inline. The PR reports 58 FE and 22 BE unit tests; I did not independently run builds or tests, and the PR states that real AWS and FE/BE end-to-end regression were not run.
- User focus: no additional focus points were supplied.
Validation was static only, as required by this review task.
| } | ||
| if (resetProgress) { | ||
| this.progress = new KinesisProgress(); | ||
| this.shardTopology.reset(); |
Contributor
There was a problem hiding this comment.
[P1] Use the full source-identity change here, not only resetProgress (which is true only when stream is supplied). The new code correctly treats region and endpoint changes as sourceChanged, but a region-only or endpoint-only ALTER retains the previous topology, completion states, explicit-shard filter, lag, and sequence numbers. Kinesis shard IDs are reused across sources, so the next task can send an old source's sequence as AFTER_SEQUENCE_NUMBER to the new source or leave a same-named shard permanently completed. Please reset all source-bound state for stream/region/endpoint changes and then apply any explicit new positions; cover region-only and endpoint-only ALTER plus journal replay.
Contributor
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
TPC-H: Total hot run time: 28384 ms |
Contributor
TPC-H: Total hot run time: 27823 ms |
Contributor
TPC-DS: Total hot run time: 152719 ms |
Contributor
ClickBench: Total hot run time: 24.13 s |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: Restore parent completion barriers so a reshard child is scheduled only after every required parent becomes visible. Stop caught-up shards from polling for the rest of a batch without reporting EOF. Remove test-only job state and synthetic transaction paths, compare and copy topology structurally, and prune expired completed nodes while retaining referenced ancestors. Skip missing jobs in Kinesis metadata replay with a warning. Limit each latest-position RPC to one scan slot and requeue after each shard so large jobs cannot occupy the whole pool. Preserve completed results when callback cleanup or FE polling crosses the deadline, while keeping cancellation and failure atomic. Add tests for scheduling fairness, queue rejection, visibility barriers, replay, topology collection and batch completion. ### Release note Improve Kinesis Routine Load reshard scheduling, idle polling and latest-position initialization. Per-job latest-position scans now run one shard at a time. ### Check List (For Author) - Test: Unit Test - ./run-be-ut.sh --run --filter='*Kinesis*' -j48: 30 passed. - ./run-fe-ut.sh --run 'org.apache.doris.load.routineload.KinesisRoutineLoadJobTest,org.apache.doris.load.routineload.kinesis.*Test': 68 passed. - No AWS or cluster regression tests run. - BE format and build hygiene checks passed. New test clang-tidy checks passed; full changed-file analysis still reports pre-existing diagnostics outside this patch. - Behavior changed: Yes (parent visibility barriers, caught-up batch completion, topology pruning and per-job scan scheduling). - Does this need documentation: No
0AyanamiRei
force-pushed
the
wip/kinesis-shard-topology-20260916
branch
from
September 28, 2026 02:20
1bd6dbf to
6ae2a75
Compare
Contributor
Author
|
run buildall |
Contributor
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
### What problem does this PR solve? Issue Number: None Related PR: apache#68357, apache#68765 Problem Summary: CI merges this branch with master before building. Sync master d437841 so local validation includes the current persistence and file-cache tests and the upstream Lance install-guard removal that restores compatibility with prebuilt compilation images. The merge has no conflict resolutions. ### Release note None ### Check List (For Author) - Test: No need to test this merge separately; targeted validation follows with the CI test fixes. - Behavior changed: No additional behavior beyond upstream master - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: After merging master, the Kinesis variant of RoutineLoadJobPersistenceTest still sets the removed openKinesisShards field through reflection and fails before exercising task scheduling. Initialize a confirmed shard topology with the existing concrete sequence position instead. Preserve the assertions that the create journal precedes publication and scheduling and that a published job can create a task. ### Release note None ### Check List (For Author) - Test: Unit Test - Before the fix, RoutineLoadJobPersistenceTest,Kinesis*Test reproduced the missing-field exception (70 tests, 1 error). - MAVEN_ARGS='-Dmaven.compiler.useIncrementalCompilation=false' ./run-fe-ut.sh --run 'RoutineLoadJobPersistenceTest,Kinesis*Test': 70 passed, no failures, errors, or skips. - mvn -f fe/pom.xml checkstyle:check -pl fe-core: 0 violations. - Behavior changed: No; test fixture only - Does this need documentation: No
### What problem does this PR solve? Issue Number: None Related PR: apache#68357 Problem Summary: ASAN in the file-cache index-only load test reports a heap-use-after-free when the main thread grows created_files while an asynchronous S3 upload callback traverses and updates its elements. Protect file creation, upload observations, and assertion readers with one fixture-owned mutex. Preserve asynchronous uploads and all existing load and compaction assertions. The fixture outlives the writers and callbacks. ### Release note None ### Check List (For Author) - Test: Unit Test - ENABLE_PCH=OFF ./run-be-ut.sh --run --filter='CloudFileCacheWriteIndexOnly*:*Kinesis*' -j48: 38 passed under ASAN. - ENABLE_PCH=OFF GTEST_REPEAT=200 ./run-be-ut.sh --run --filter='CloudFileCacheWriteIndexOnlyTest.*' -j48: all 200 iterations passed (600 executions). - The unpatched local baseline passed 20 repetitions; the concurrency failure was observed in CI, not reproduced locally. - clang-format 16 is clean for the modified file; build hygiene and clang-tidy for modified lines passed. - Behavior changed: No; test observations only - Does this need documentation: No
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 29518 ms |
Contributor
TPC-DS: Total hot run time: 151963 ms |
Contributor
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
TPC-H: Total hot run time: 28961 ms |
Contributor
TPC-DS: Total hot run time: 152229 ms |
Contributor
ClickBench: Total hot run time: 25.69 s |
### What problem does this PR solve? Related PR: apache#68357 Problem Summary: The Clang Formatter check fails because the Kinesis upstream mock test has incorrectly ordered includes and noncanonical argument wrapping. Apply clang-format 16 to the test file without changing test behavior. ### Release note None ### Check List (For Author) - Test: Manual test - build-support/check-format.sh (passed with clang-format 16.0.6) - build-support/check-build-hygiene.sh (passed) - git diff --check (passed) - No unit test rerun locally: formatting-only change - Behavior changed: No - Does this need documentation: No
Contributor
Author
|
run buildall |
Contributor
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
TPC-H: Total hot run time: 29167 ms |
Contributor
TPC-DS: Total hot run time: 151757 ms |
Contributor
FE Regression Coverage ReportIncrement line coverage |
Contributor
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
TPC-H: Total hot run time: 29210 ms |
Contributor
TPC-DS: Total hot run time: 152696 ms |
Contributor
ClickBench: Total hot run time: 25.62 s |
liaoxin01
approved these changes
Oct 14, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: None
Related PR: #67678, #67696, #67869, #67874, #67879
Kinesis shards can split and merge while Routine Load is running. Open/closed shard lists and sequence progress alone do not durably identify the initial shard snapshot or represent parent completion dependencies. This can skip new-shard backlog, reconsume completed parents, move LATEST positions when tasks are recreated, or process child data before the parent becomes visible.
This PR introduces a persisted shard topology and lifecycle:
The change also incorporates ALTER validation atomicity, FE/BE tests, an in-process AWS client fixture, and a LATEST restart regression suite.
Release note
Improve Kinesis Routine Load shard discovery, durable starting positions and progress across split/merge. Child scheduling waits for required parents to become visible. Idle polling is reduced, and latest-position initialization shares BE scan capacity across jobs.
Check List (For Author)
./run-fe-ut.sh --run 'org.apache.doris.load.routineload.KinesisRoutineLoadJobTest,org.apache.doris.load.routineload.kinesis.*Test'— 68 passed, no failures/errors/skips../run-be-ut.sh --run --filter='*Kinesis*' -j48— 30 passed, no failures/errors/skips.mvn checkstyle:check -pl fe-core— 0 violations.Implementation limits: topology journals still contain complete retained ListShards snapshots when changes occur. Scan fairness is at shard boundaries, with one shard scanned at a time per job; page-level preemption and load-aware BE selection are not implemented.