Skip to content

[fix](be) Continue Kinesis polling after empty records - #67874

Closed
0AyanamiRei wants to merge 2 commits into
apache:masterfrom
0AyanamiRei:fix/kinesis-empty-page-r2
Closed

0AyanamiRei wants to merge 2 commits into
apache:masterfrom
0AyanamiRei:fix/kinesis-empty-page-r2

Conversation

@0AyanamiRei

Copy link
Copy Markdown
Contributor

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:

  1. KinesisDataConsumer::group_consume() called GetRecords.
  2. It saved the returned NextShardIterator.
  3. When Records was empty, it erased the shard from _consuming_shard_ids.
  4. If this was the only active shard, the loop marked the consumer done,
    shut down the queue, and returned successfully.
  5. The next task recreated an iterator from the old committed sequence, so it
    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:

  • response 1: no records, MillisBehindLatest = 100, iterator
    iterator-after-empty-page;
  • response 2: one record, MillisBehindLatest = 0, empty iterator.

Before this fix, the test failed because the consumer made only one
GetRecords call:

Expected: 2
Actual:   1

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)

  • Test: Unit Test
    • ./run-be-ut.sh --run --filter='KinesisEmptyPageReproduction.*' -j48
      failed before the fix and passed 2/2 after the fix.
    • ./run-be-ut.sh --run --filter='KinesisBatchProgressReproduction.:KinesisEmptyPageReproduction.' -j48
      passed 7/7 after the fix, including the existing Kinesis progress
      boundary tests from the base branch.
    • build-support/check-build-hygiene.sh passed.
    • build-support/check-format.sh passed.
    • git diff --check passed.
    • The new test's clang-tidy check passed. The production file's full
      clang-tidy check still reports pre-existing diagnostics outside this
      change, including existing complexity and const-reference warnings.
  • Behavior changed: Yes. Empty non-terminal Kinesis responses are polled
    again using the returned iterator.
  • Does this need documentation: No.

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.

### 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
@hello-stephen

Copy link
Copy Markdown
Contributor

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

  1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
  2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
  3. What features were added. Why was this function added?
  4. Which code was refactored and why was this part of the code refactored?
  5. Which functions were optimized and what is the difference before and after the optimization?

@0AyanamiRei
0AyanamiRei marked this pull request as draft September 15, 2026 19:08
@0AyanamiRei

Copy link
Copy Markdown
Contributor Author

draft,计划是希望对routine load kinese先去持久化的维护一个shards父子关系的拓扑图

@0AyanamiRei

Copy link
Copy Markdown
Contributor Author

Absorbed into the Kinesis shard topology change, closing this draft.

  • data_consumer.cpp: an empty GetRecords response no longer ends the round; the shard stays in the active set and the returned NextShardIterator is followed, exactly as in this PR (the surrounding lines were rewritten by the queue/marker change, so the hunk is not textually identical).
  • be/test/runtime/kinesis_empty_page_test.cpp: added, with the queue type adapted from BlockingQueue<std::shared_ptr<Record>> to BlockingQueue<KinesisQueueItem> and the assertion moved to item.record. That is the interface mismatch anticipated when this PR is stacked on the queue change; it is resolved here.

An empty page is now also what makes the "shard has not reached its tip" case explicit in the state machine: only an empty NextShardIterator produces the end-of-shard marker.

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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants