0AyanamiRei opened a new pull request, #67869:
URL: https://github.com/apache/doris/pull/67869
### 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:
- Queue items carry the shard ID along with the record.
- The consumer group updates candidate progress only after a successful pipe
append and uses that progress in the existing transaction attachment.
- End-of-shard markers are queued after the shard's records. The group
reports completion only after consuming the marker.
- Queue shutdown while enqueueing a prefetched response is treated as normal
batch termination. Pipe and source errors continue through the existing task
failure paths.
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)
- Test:
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test
- [ ] No need to test
- Behavior changed:
- [ ] No.
- [x] Yes. Task progress is limited to successfully appended records,
and shard completion follows an ordered queue marker.
- Does this need documentation:
- [x] No. No new configuration or SQL syntax.
- [ ] Yes.
Validation:
```bash
./run-be-ut.sh --run --filter='KinesisBatchProgressReproduction.*' -j48
```
- Before the fix: the full-batch control passed; the row-limit and
byte-limit cases failed with expected sequence 2 and actual sequence 5.
- After the fix: all 5 ASAN BE unit tests passed. Coverage includes
full-batch progress, row and byte limits, processing a shard-end marker after
records, and shutdown while a producer is blocked on a full queue.
- The tests invoke the real producer record-processing and consumer-group
methods with constructed SDK records and a recording pipe.
- Header hygiene, clang-format checks, and `git diff --check` passed.
- Live AWS-to-Doris ingestion and time-limit boundary tests were not run.
- Clang-tidy was attempted but did not pass due to pre-existing diagnostics;
no clean clang-tidy result is claimed. The existing polling function's size and
cognitive-complexity suppressions are documented inline.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
