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:

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.

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

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

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, GetRecords.

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
  • Unit Test
  • Manual test (real AWS/network cancellation was not exercised)
  • No need to test or manual test
• Behavior changed:
  • No.
  • Yes. Initial OPEN LATEST shards are scanned before scheduling, complete results are journaled, and PAUSE cancels pending initialization.
• Does this need documentation?
  • No.
  • Yes. New settings and initialization semantics are described above; a website documentation PR is not included.

Validation performed

Commands used (the local Arrow 24/Paimon installation uses the unversioned prefix):

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