Repository navigation
[fix](be) Continue Kinesis polling after empty records - #67874
Closed
0AyanamiRei wants to merge 2 commits into
Closed
0AyanamiRei wants to merge 2 commits into
0AyanamiRei wants to merge 2 commits into
Conversation
### What problem does this PR solve? Issue Number: N/A Related PR: apache#61325 Problem Summary: Kinesis GetRecords can return an empty Records array while the returned NextShardIterator still points to a later position. The routine load consumer treated every such response as the end of the current consumption round and removed the shard from its active set. When the shard was the only active shard, the task ended immediately, discarded the new iterator at task reset, and the next task recreated an iterator from the old committed sequence. Records that became available after the empty response could therefore be delayed indefinitely and, for a task that kept restarting from the same position, never consumed. Keep the shard active when Records is empty and NextShardIterator is non-empty. The consumer now follows the returned iterator on the next polling round. A null/empty NextShardIterator continues to use the existing end-of-shard path, so closed-shard markers, queue ordering, transaction attachments, and FE progress handling remain unchanged. ### Release note Fix Kinesis Routine Load polling after empty GetRecords responses so records that arrive later in the same shard are not skipped. ### Check List (For Author) - Test: Unit Test - Exact command: ./run-be-ut.sh --run --filter='KinesisEmptyPageReproduction.*' -j48 - Before the fix, the test failed with expected 2/3 GetRecords calls and actual 1. - After the fix, the same test passed 2/2. - Combined R1/R2 command passed 7/7: ./run-be-ut.sh --run --filter='KinesisBatchProgressReproduction.*:KinesisEmptyPageReproduction.*' -j48 - Tests call production KinesisDataConsumer::group_consume with AWS SDK fake responses and verify that NextShardIterator is used. - Behavior changed: Yes (empty non-terminal responses are polled again) - Does this need documentation: No
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
0AyanamiRei
marked this pull request as draft
September 15, 2026 19:08
Contributor
Author
|
draft,计划是希望对routine load kinese先去持久化的维护一个shards父子关系的拓扑图 |
Contributor
Author
|
Absorbed into the Kinesis shard topology change, closing this draft.
An empty page is now also what makes the "shard has not reached its tip" case explicit in the state machine: only an empty |
0AyanamiRei
added a commit
to 0AyanamiRei/doris
that referenced
this pull request
Sep 23, 2026
### 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
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: N/A
Related PR: #67869
Problem Summary:
Amazon Kinesis GetRecords is a non-blocking pull API. An empty Records array
does not mean that the shard has reached the stream tip: the response can
still contain a non-empty NextShardIterator, and a later call using that
iterator can return records. The only terminal signal for a closed shard is
an empty NextShardIterator.
Routine Load Kinesis handled the response as follows:
shut down the queue, and returned successfully.
could observe the same empty page again instead of following the iterator
returned by Kinesis.
This can leave a shard stuck without consuming records that arrive after the
empty response. It also makes the result depend on whether the task happens
to have another active shard.
This change keeps the shard in the active set when Records is empty and
NextShardIterator is non-empty. The next polling round therefore follows the
returned iterator. The existing empty-iterator branch is unchanged and
continues to remove the closed shard only after the existing queue ordering
and end-of-shard handling. Queue shutdown, retry handling, transaction
attachments, and FE progress formats are unchanged.
Reproduction
The new BE unit test invokes the production
KinesisDataConsumer::group_consume() method with an AWS SDK KinesisClient
fake:
iterator-after-empty-page;
Before this fix, the test failed because the consumer made only one
GetRecords call:
After this fix, it verifies both that the second call is made and that it
uses iterator-after-empty-page. A second test covers two consecutive empty
responses before the record.
Release note
Fix Kinesis Routine Load polling after empty GetRecords responses so later
records in the same shard are not skipped or left permanently unconsumed.
Check List (For Author)
failed before the fix and passed 2/2 after the fix.
passed 7/7 after the fix, including the existing Kinesis progress
boundary tests from the base branch.
clang-tidy check still reports pre-existing diagnostics outside this
change, including existing complexity and const-reference warnings.
again using the returned iterator.
Scope and limitations
This PR addresses only empty non-terminal GetRecords responses. It does not
change shard discovery, reshard parent/child ordering, LATEST initialization,
or FE shard-topology persistence. Live AWS ingestion was not used because the
machine has multiple shared Doris processes and no verifiable cluster identity
for this workspace; the deterministic unit test exercises the production BE
consumer and AWS SDK response path without creating external resources.