[spoon-lz]

Purpose

Fixes #4605.

Preserve successfully decoded Arrow batches when a later batch fails to decode during the same poll.

This aligns the partial-result delivery policy with Java's Arrow batch scanner and prevents subscription offsets from advancing past successful data that was never delivered to the caller.

Brief change log

• Retain a decoding error after collecting successful batches and report it on the next poll.
• Preserve partial results both within one completed fetch and across completed fetches.
• Report errors immediately when no successful batches are available.
• Use existing subscription and offset checks to discard stale deferred errors.
• Document the batch poll behavior and add regression tests.

Tests

• The same collector regression fails on the community baseline with an Arrow parse error and passes with this change.
• Rust workspace unit-test gate: 855 tests passed.
• Two final regression tests passed after test refactoring; the collector test covers 18 scenarios.
• Workspace Clippy and Rustdoc with warnings denied: passed.
• Formatting and diff checks: passed.

Coverage includes immediate errors, multiple successful batches, same-fetch and cross-fetch failures, empty fetches, offset continuity, recovery, unsubscribe, and subscription offset changes.

No live Fluss-cluster, OSS, or Python/C++ runtime integration tests were run.

API and Format

No public API signature, configuration, protocol, or storage-format changes.

Arrow batch poll error ordering changes: successfully decoded batches are returned before a later decoding error. Fetch-response initialization and authorization errors remain outside this change.

Documentation

Updated the Rust RecordBatchLogScanner::poll API comment to describe partial-result delivery and deferred errors.

Generative AI disclosure

Yes — Codex was used to prepare the implementation, tests, and documentation.

[naivedogger]

Thanks @spoon-lz for this PR. Left some comments, PTAL.

[naivedogger]

If a valid batch from bucket A is followed by an authorization error from bucket B, the initialize_fetch() error branch returns an error after A's offset has advanced, without delivering A's batch. Could we return the collected batches first, retain the error for the next poll, and add a regression test for this path?

[spoon-lz]

Fixed. Collected batches are now returned first, with the initialization error deferred to the next poll. Added a cross-bucket authorization-error regression test.

[naivedogger]

Thanks for the update. There seems to be one remaining case: remote download failures are wrapped with offset -1 in LogFetchBuffer::try_complete(). If deferred after valid batches, the error is discarded on the next poll because fetch_batches_from_fetch() treats the offset mismatch as a stale response.

Could we preserve the original fetch offset for these failures and add a regression test with valid data followed by a failed pending fetch?

[spoon-lz]

Fixed by preserving the original fetch offset when a pending fetch fails. Added regression tests verifying that valid batches are returned first and the download error is reported on the next poll.

[naivedogger]

Could we also test an uninitialized fetch with an authorization error and no preceding data? The result.is_empty() && size_in_bytes == 0 check currently discards the error and returns an empty result. Please propagate the error immediately; calling set_initialized() here bypasses that path.

[spoon-lz]

Fixed. Zero-byte authorization errors are now propagated immediately when no data is available. Added a regression using an uninitialized fetch, without calling set_initialized().

[naivedogger]

LGTM. Thanks @spoon-lz for the updates. cc @loserwang1024