Repository navigation
[fix](be) Commit Kinesis progress only for appended records - #67869
0AyanamiRei wants to merge 1 commit into
Conversation
### What problem does this PR solve? Issue Number: N/A Related PR: apache#61325 Problem Summary: Kinesis Routine Load advanced a shard's progress while records were only prefetched into the BE queue. When a task reached its row or byte batch limit, the Group could append only a prefix of that queue, while the transaction attachment carried the producer's later sequence number. The next task then resumed after records that had never been loaded. The fix associates each queued record with its shard, records candidate progress only after a successful Pipe append, and orders closed-shard markers after their records. Queue shutdown during a batch boundary now finishes the prefetched producer without promoting unappended records. ### Release note Fix Kinesis Routine Load skipping prefetched records at row or byte batch boundaries. ### Check List (For Author) - Test: Unit Test - `./run-be-ut.sh --run --filter='KinesisBatchProgressReproduction.*' -j48` - 5 tests passed after the fix; the row and byte boundary cases failed before the fix with committed sequence 5 versus appended sequence 2. - Behavior changed: Yes. Kinesis task progress and closed-shard reporting now follow successfully appended queue items. - Does this need documentation: No.
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
sollhui
left a comment
sollhui
left a comment
There was a problem hiding this comment.
Found a consumer-exit issue in the graceful queue-shutdown path; details inline.
| if (!queue->controlled_blocking_put(item, config::blocking_queue_cv_wait_timeout_ms)) { | ||
| // The group may have reached a batch boundary while this consumer was still draining | ||
| // a prefetched response. The appended prefix remains a valid batch. | ||
| return Status::OK(); |
There was a problem hiding this comment.
[P2] Propagate queue shutdown so the whole consumer exits successfully
Returning OK here makes group_consume() treat an interrupted enqueue as a fully processed response. If a consumer owns shards A, B and C, and the group reaches its batch limit while the producer is enqueueing A's response, the group shuts down the queue, cancels the consumer and waits in _thread_pool.join(). This helper now returns OK, so group_consume() continues its inner shard loop and issues GetRecords for B and C even when _cancelled is already true; cancellation and the remaining time budget are only checked in the outer loop.
Consequently, completing an otherwise valid batch waits for unnecessary AWS calls, potentially exceeding the task/transaction timeout with many shards or slow requests. A source error from one of those extra calls can also turn the batch into a failure and roll back the appended prefix.
I reproduced this with the unchanged group_consume() and _process_records() methods from this commit in an isolated C++ harness, with AWS/queue infrastructure stubbed. When A's enqueue returns false with cancellation already set, the current code calls GetRecords for A, B and C. Restoring the original error return as a control stops after A. This is not a full Doris/AWS integration test.
Please propagate an explicit queue-shutdown/stop indication from _process_records() and have group_consume() return OK immediately for that condition, preserving graceful batch completion. Checking cancellation before each shard request would also help. The existing QueueShutdownDuringPrefetchIsGraceful test only calls the helper, so please add a multi-shard test that asserts no subsequent source requests occur after shutdown.
|
draft，计划是希望对routine load kinese先去持久化的维护一个shards父子关系的拓扑图 |
|
Absorbed into the Kinesis shard topology change, closing this draft. The fix landed in the working branch (
The commit boundary introduced here is the BE half of the new state machine: the FE reads |
### 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, 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: N/A
Related PR: #61325
Problem Summary:
Kinesis Routine Load can skip records when a task reaches its batch limit while prefetched records remain in the consumer queue.
The producer previously advanced its per-shard sequence number in
_process_records()before enqueueing each record. The consumer group then copied that producer progress into the transaction attachment, even when it had only appended part of the queue to the Stream Load pipe.For example, a producer queues records S1 through S5, but the group reaches its row or byte limit after appending S1 and S2. Previously, the task reported S5 as its commit position. If the transaction succeeded, the next task resumed with
AFTER_SEQUENCE_NUMBER(S5), skipping S3 through S5 even though those records had never entered the completed load.The same boundary affects shard completion: observing the end of a shard in the producer does not prove that all of its queued records have been appended. Reporting the shard as closed too early can remove its remaining records from FE tracking.
This PR fixes the progress boundary:
After the change, the example task reports S2, so subsequent consumption starts after S2 and can load S3 through S5. Pipe append establishes candidate progress; the existing transaction path still determines whether that progress is committed.
The FE/BE wire protocol and persisted progress format are unchanged.
Release note
Fix Kinesis Routine Load skipping prefetched records at batch boundaries and reporting shard completion before queued records have been appended.
Check List (For Author)
Validation:
./run-be-ut.sh --run --filter='KinesisBatchProgressReproduction.*' -j48git diff --checkpassed.Check List (For Reviewer who merge this PR)