Skip to content

[fix](fe) Preserve Kinesis initial shard position state - #67879

Closed
0AyanamiRei wants to merge 1 commit into
apache:masterfrom
0AyanamiRei:fix/kinesis-initial-position-r4
Closed

[fix](fe) Preserve Kinesis initial shard position state#67879
0AyanamiRei wants to merge 1 commit into
apache:masterfrom
0AyanamiRei:fix/kinesis-initial-position-r4

Conversation

@0AyanamiRei

@0AyanamiRei 0AyanamiRei commented Sep 11, 2026

Copy link
Copy Markdown
Contributor

What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary:

Kinesis Routine Load used an empty progress map as the signal that a job was undergoing its initial shard setup. This is not a valid lifecycle invariant. When a closed parent shard is fully consumed, KinesisProgress.update() removes that shard from the progress map. The map can therefore become empty after the job has already completed its initial positioning.

A reshard can create child shards before the next FE metadata refresh. In that window, KinesisRoutineLoadJob.updateNewShardProgress() sees an empty progress map and treats the children as initial shards. It applies the configured LATEST position. Kinesis LATEST starts after the current tip, so records written to the children before FE discovers them are skipped.

The fix stores this lifecycle state separately from the currently tracked shard positions:

  • New jobs start with kinesisInitialPositionSet = false.
  • After initial shard positions are established, the marker is set to true.
  • A committed or replayed Kinesis task also confirms that initial positioning has happened, so the marker survives parent-shard progress removal.
  • Newly discovered shards after initialization continue to use TRIM_HORIZON, even when the progress map is empty.
  • The marker is persisted in the Kinesis routine load job image.
  • Legacy images without the marker infer the state from the persisted job state and shard lists.
  • ALTER that changes the stream and resets progress clears the marker so the new stream uses its configured initial position.

The test directly invokes the production FE method and models the complete boundary: initial shard setup with LATEST, removal of the completed parent from progress, and discovery of a child shard. It verifies that the child receives TRIM_HORIZON. Additional assertions cover transaction progress handling, image serialization, and stream-change reset.

Release note

Prevent Kinesis Routine Load from skipping records on newly discovered child shards after parent shard progress is exhausted.

Check List (For Author)

  • Test
    • Regression test
    • Unit Test
      • ./run-fe-ut.sh --run org.apache.doris.load.routineload.KinesisRoutineLoadJobTest
      • 10 tests passed, 0 failures, 0 errors, 0 skipped
      • The same test failed before the fix with: expected: <-2> but was: <LATEST>
    • Manual test
    • No need to test or manual test
  • Additional validation:
    • cd fe && mvn checkstyle:check -pl fe-core -Dcheckstyle.skip=false passed
    • build-support/check-build-hygiene.sh passed
    • build-support/check-format.sh passed
    • git diff --check passed
  • Behavior changed:
    • No.
    • Yes. Initial shards retain the configured default position, while shards discovered after initial setup use TRIM_HORIZON.
  • Does this need documentation:
    • No.
    • Yes.

The FE unit test was chosen because R4 is a frontend lifecycle-state bug. It calls the production KinesisRoutineLoadJob implementation and avoids a timing-dependent AWS integration setup. No AWS credentials were read and no Kinesis stream was created.

This PR is based on origin/master and contains only the R4 implementation and its FE unit-test coverage. Existing Kinesis BE consumer fixes remain separate.

Check List (For Reviewer who merge this PR)

  • Confirm the release note
  • Confirm test cases
  • Confirm document
  • Add branch pick label

@0AyanamiRei
[fix](fe) Preserve Kinesis initial shard position state
9f951ab 
### What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary: Kinesis Routine Load used an empty progress map as the signal that a job was undergoing its initial shard setup. This is not a valid lifecycle invariant: after a closed parent shard is fully consumed, KinesisProgress removes that shard, and the progress map can become empty even though the job has already completed its initial positioning.

If a reshard creates a child shard before the next FE refresh, updateNewShardProgress() sees an empty progress map and treats the child as an initial shard. It applies the configured LATEST position, which starts after the current tip and skips records written to the child before FE discovers it. The same state could be lost across image recovery because the previous implementation had no durable job-level initialization marker.

Keep initial-position state separate from the current set of tracked shards. New jobs start with the marker unset, and the marker is set after initial shard positions are established or a Kinesis task transaction is processed. Later shards therefore start at TRIM_HORIZON even when all prior shard progress has been removed. The marker is serialized in the Kinesis routine load job image, recovered for legacy images from their persisted lifecycle state, and reset when ALTER changes the stream and clears progress.

### Release note

Prevent Kinesis Routine Load from skipping records on newly discovered child shards after parent shard progress is exhausted.

### Check List (For Author)

- Test: Unit Test
    - `./run-fe-ut.sh --run org.apache.doris.load.routineload.KinesisRoutineLoadJobTest`
    - 10 tests passed, 0 failures, 0 errors, 0 skipped
    - `build-support/check-build-hygiene.sh` passed
    - `build-support/check-format.sh` passed
    - `git diff --check` passed
- Behavior changed: Yes. Initial shards preserve the configured default position; shards discovered after initial setup use TRIM_HORIZON.
- Does this need documentation: No. This fixes the existing Kinesis shard lifecycle behavior.

The unit test directly invokes the production FE updateNewShardProgress() path. It models initial shard positioning, removal of the completed parent from progress, and discovery of a child shard, then verifies that the child uses TRIM_HORIZON. No AWS resource is created by this test.
@hello-stephen

hello-stephen commented Sep 11, 2026

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

0AyanamiRei commented Sep 15, 2026

Copy link
Copy Markdown
Contributor Author

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

@0AyanamiRei

0AyanamiRei commented Sep 21, 2026

Copy link
Copy Markdown
Contributor Author

Superseded by the Kinesis shard topology change, closing this draft without merging it as-is.

The problem this PR fixes is real and is covered, but the marker it introduces would duplicate topology state:

  • This PR adds a job-level kinesisInitialPositionSet flag so that an empty progress map is no longer read as "first setup".
  • The topology change replaces that signal entirely: KinesisShardTopology.initialSnapshotFinalized freezes the initial shard snapshot, and each node keeps its own initialStartPosition. A shard discovered after the snapshot gets TRIM_HORIZON; the initial snapshot keeps the configured default position. Neither decision consults the progress map, so "progress became empty because the consumed parent was dropped" can no longer change the outcome.
  • The child-lineage path from GetRecords.ChildShards resolves to the same TRIM_HORIZON.

The regression is covered by the new tests:

  • KinesisShardTopologyTest.testShardDiscoveredAfterInitialSnapshotUsesTrimHorizon (parent consumed, children discovered afterwards, configured default is LATEST),
  • KinesisShardTopologyTest.testChildShardLineageUsesTrimHorizonAfterParentCompletion (lineage path, still waiting for the parent's visibility barrier),
  • KinesisRoutineLoadJobTest.testShardDiscoveredAfterInitialSnapshotUsesTrimHorizonInProgress (the same scenario at job level, asserting the position that reaches progress).

@0AyanamiRei 0AyanamiRei mentioned this pull request Sep 22, 2026
Closed
0AyanamiRei added a commit to 0AyanamiRei/doris that referenced this pull request Sep 23, 2026
@0AyanamiRei
[feature](load) Add Kinesis shard topology scheduling
4a2deb8 
### 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

Reviewers

@liaoxin01 liaoxin01 Awaiting requested review from liaoxin01 liaoxin01 is a code owner
@gavinchou gavinchou Awaiting requested review from gavinchou gavinchou is a code owner

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@0AyanamiRei @hello-stephen