0AyanamiRei opened a new pull request, #67879:
URL: https://github.com/apache/doris/pull/67879
### 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 <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] 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.
- [x] Yes. Initial shards retain the configured default position, while
shards discovered after initial setup use `TRIM_HORIZON`.
- Does this need documentation:
- [x] 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
--
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]
