Skip to content

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

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

[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

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

### 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

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?

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