0AyanamiRei opened a new pull request, #67678:
URL: https://github.com/apache/doris/pull/67678
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Kinesis Routine Load can skip records when a shard starts at `LATEST` and
its first task does not commit a concrete sequence number. The BE has resolved
`LATEST` into a shard iterator, but FE progress still contains the symbolic
position. Empty polling, an aborted transaction, a timeout, or task replacement
can discard the iterator. The replacement task then interprets `LATEST` again
against a later stream tip.
For example:
```text
1. A shard contains R1, R2, R3; task A obtains a LATEST iterator after R3.
2. R4 and R5 arrive.
3. Task A aborts or is replaced before committing a sequence number.
4. FE still supplies LATEST to task B.
5. Task B obtains an iterator after R5, skipping R4 and R5.
```
Kafka resolves symbolic offsets through its watermark metadata API before
consumption. Kinesis `GetShardIterator(LATEST)` returns an opaque iterator, not
a durable numeric end offset, and an empty `GetRecords` response does not
reveal the preceding record's sequence number. The five-minute expiration
applies to the iterator; a record sequence number can be used to obtain another
iterator while the source position remains available within retention.
This PR resolves initial Kinesis positions before creating data tasks,
persists the resolved positions in FE, and keeps the existing task and
transaction progress protocol.
#### 1. Resolve only the shards that need initialization
FE selects the intersection of the job's currently tracked **OPEN shards**
and progress entries still set to `LATEST` (case insensitive) or `-1`.
- Concrete sequence numbers and `TRIM_HORIZON` positions are reused without
scanning.
- Shards already known to be CLOSED do not participate in initialization
scanning; existing closed-shard consumption/cleanup remains in place.
- Stale progress entries left after reducing a custom shard selection do not
trigger unnecessary scans.
- Existing shard discovery supplies the OPEN/CLOSED tracking; this change
does not add discovery calls just to determine that state.
`prepare()` submits one asynchronous metadata RPC containing the batch. The
job remains `NEED_SCHEDULE` while the result is pending.
`divideRoutineLoadJob()` checks the same unresolved set and does not create
tasks until resolution and persistence have completed. No data-task transaction
is opened while waiting for this scan.
#### 2. Find restartable positions by scanning retained records
For each selected shard, the BE requests a `TRIM_HORIZON` iterator and
follows `GetRecords` pages, keeping the last record's sequence number. It
continues across empty intermediate pages and finishes when the service reports
`MillisBehindLatest == 0` or the iterator reaches a closed shard's end.
| Scan result | Persisted FE position | Subsequent task behavior |
|---|---|---|
| The shard tip is R3 | `sequence(R3)` | Read with
`AFTER_SEQUENCE_NUMBER(R3)` |
| The scan reaches the tip without seeing any records | `TRIM_HORIZON` |
Read the earliest retained records, including later arrivals |
| A shard fails, the batch times out, or initialization is cancelled | No
positions from that batch are published | Keep initialization unresolved |
The scan downloads historical payloads but never sends them to the Doris
load pipe. Its total work is `O(sum(L_i))`, or `O(N * L)` for N shards with
average retained length L. Each worker retains a response page rather than
buffering the entire history.
#### 3. Bound concurrency across all initialization requests on a BE
A dedicated BE scan pool is shared by all batches, with **8 workers by
default**. Each batch submits at most `min(shard_count,
configured_worker_count)` workers. Workers claim shards within their batch, and
each worker owns its scan consumer/client.
Scanning is dispatched directly to this pool, so ordinary Routine Load
workers do not block waiting for long metadata scans. The queue is bounded to
1024 workers; submission failure fails the batch and causes already submitted
workers to stop. This is a concurrency bound, not a guarantee of fair
scheduling between jobs.
The result map is protected by a mutex. Completion accounting includes both
executed workers and failed submissions. The response and RPC `done` callback
are completed only after all workers have exited, so no worker can continue
using the RPC controller after it is released.
#### 4. Persist the complete batch before scheduling tasks
```text
FE receives the complete successful response
-> verify RPC status and requested shard coverage
-> synchronously journal KinesisLatestPositionOperation(jobId, positions)
-> apply positions to existing KinesisProgress
-> create tasks from that progress
-> begin transactions and use the existing task protocol
```
The new operation is wired into journal serialization and replay. The
existing job image continues to persist the progress map. A crash after
journaling but before applying the in-memory update recovers the positions
through replay.
No successful subset is persisted or reused after a failed initialization
attempt. Once the batch is persisted, an abort or replacement uses the saved
position instead of choosing a newer tip. Successful data transactions continue
advancing progress through the original attachment path; no changes are made to
the data-task Thrift schema or transaction commit behavior.
#### 5. Make cancellation visible to the BE
The current brpc implementation does not reliably expose cancellation of an
individual gRPC stream through `Controller::IsCanceled()`, which observes
socket failure. Each initialization batch therefore uses its own gRPC channel,
created through the existing channel provider and DNS handling. A Future
listener calls non-blocking `shutdownNow()` on completion, failure, or
cancellation.
PAUSE, terminal job states, explicit source ALTER, and an enabled total
timeout cancel the pending initialization. Closing its dedicated connection
lets the BE observe cancellation without closing connections used by unrelated
requests. This does not add a cancel RPC or a scan-registration handshake.
Workers check shared batch status before claiming shards, around AWS calls,
and during backoff. AWS requests also receive a continuation callback. The BE
checks server shutdown as well, preventing an unlimited scan from blocking brpc
shutdown before the executor's stop phase.
#### Configuration and timeout semantics
| Setting | Location | Default | Meaning |
|---|---|---|---|
| `kinesis_latest_sequence_timeout_second` | FE, mutable | `-1` | Total
initialization budget; -1 disables it, a positive value is seconds |
| `kinesis_latest_sequence_scan_threads` | BE, startup | `8` | Maximum
concurrent scan workers shared by that BE |
| `kinesis_latest_sequence_request_timeout_ms` | BE, startup | `30000` |
Finite total timeout for each scan-client Kinesis HTTP request |
A positive overall timeout covers the batch, including BE queueing, and is
never reset per shard or successful page. FE also sets a gRPC deadline and
rejects results collected after its deadline. BE workers share one local
monotonic deadline established before queueing.
Scan clients use the SDK's actual HTTP timeout (`httpRequestTimeoutMs`),
independently of the ordinary consumers' settings. SDK-internal retries are
disabled for these clients; iterator acquisition and record reads use up to
three cancellable retries with 1, 2, and 4 second backoffs.
Cancellation is cooperative: the SDK continuation handler does not
immediately interrupt every silent network wait. An in-flight Kinesis HTTP call
may take until its finite request timeout to finish. Credential-provider
initialization/refresh is not fully covered by that Kinesis HTTP timeout.
#### Semantics and compatibility boundaries
- Each shard's initial boundary is the tip observed when its scan catches
up. Records encountered during initialization belong to the skipped initial
prefix. This is not a CREATE-time cutoff or a cross-shard snapshot.
- All-or-nothing persistence avoids retaining some shards' positions across
later failed attempts; it does not make the observations within one scan
simultaneous. With an unlimited total timeout, their time difference has no
fixed upper bound.
- Source retention still limits recovery; this change does not recover data
already skipped by an older version or expired from Kinesis.
- An older BE that ignores the new request field returns no position map,
which the FE rejects instead of scheduling unresolved positions.
- The new journal operation must be understood by replaying FEs.
Mixed-version upgrade/downgrade procedures must account for it.
- The ordinary Kinesis consume loop, EFO, and the data-task transaction
protocol are outside this patch.
AWS API references:
[GetShardIterator](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html),
[GetRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html).
### Release note
Fix Kinesis Routine Load initialization from `LATEST` losing its starting
position after empty polling, transaction aborts, timeouts, or task
replacement. Initial positions are resolved and journaled before data tasks are
created, using bounded BE scan concurrency.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (real AWS/network cancellation was not exercised)
- [ ] No need to test or manual test
- Behavior changed:
- [ ] No.
- [x] Yes. Initial OPEN `LATEST` shards are scanned before scheduling,
complete results are journaled, and PAUSE cancels pending initialization.
- Does this need documentation?
- [ ] No.
- [x] Yes. New settings and initialization semantics are described
above; a website documentation PR is not included.
#### Validation performed
| Check | Result |
|---|---|
| FE build/package with UI disabled | Passed |
| `KinesisLatestPositionTest` | 8 passed |
| `KinesisDataSourcePropertiesTest` | 4 passed |
| BE ASAN_UT build/link and `KinesisLatestSequenceTest.*` | 7 passed;
rebuilt and rerun after the scan-helper cleanup |
| Existing `RoutineLoadTaskExecutorTest.*` | 1 passed |
| clang-format 16, BE header hygiene, Java Checkstyle, diff whitespace
checks | Passed |
Commands used (the local Arrow 24/Paimon installation uses the unversioned
prefix):
```bash
DISABLE_BUILD_UI=ON BUILD_TYPE=ASAN ./build.sh --fe -j48
./run-fe-ut.sh --run 'org.apache.doris.load.routineload.kinesis.*Test'
ARROW_HOME="$PWD/thirdparty/installed"
PAIMON_HOME="$PWD/thirdparty/installed" \
./run-be-ut.sh --run --filter='KinesisLatestSequenceTest.*' -j48
ARROW_HOME="$PWD/thirdparty/installed"
PAIMON_HOME="$PWD/thirdparty/installed" \
./run-be-ut.sh --run --filter='RoutineLoadTaskExecutorTest.*' -j48
```
Validation limitations:
- The separate production BE build was blocked by the build script's
required versioned/dual-version thirdparty layout. The successful BE build
above is the **ASAN_UT** build, not a production BE package.
- Full clang-tidy did not pass because of an existing unmatched `NOLINTEND`
in `be/src/core/types.h` and existing whole-file warnings. The newly added
functions' complexity and request-copy diagnostics were fixed. A focused rerun
of the two adjusted C++ files reported no diagnostics on their final changed
lines; this is not a claim that full clang-tidy passed.
- UTs use mocked Kinesis responses. Real AWS behavior, cross-job concurrency
limits under load, connection-close cancellation latency, and shutdown under
stalled network I/O have not been integration-tested.
### 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]
