[flink] Advance idle bucket baselines after successful Undo Recovery #4952

Search before asking

• I searched the existing issues and found no matching report.

Fluss version

main (development), inspected at 7392085.

Please describe the bug

After successfully restoring an aggregation sink with Undo Recovery, the operator initializes its bucket offsets from the original checkpoint baseline. The recovery writer appends compensating changes, but its returned offsets are not propagated to the operator.

If a recovered bucket receives no subsequent sink writes, later successful checkpoints keep storing the old baseline. A later restore unnecessarily replays already completed recovery history and can fail with LogOffsetOutOfRangeException once log retention removes that history.

Reproduction:

1. Write an aggregation table and checkpoint the sink at bucket offset B.
2. Write additional records and restore the checkpoint, allowing Undo Recovery to complete.
3. Complete another checkpoint without further writes to that bucket.
4. Restore the new checkpoint. Its baseline is still B rather than the boundary after the compensating writes. If the old history has expired, recovery cannot start.

Expected: the next checkpoint records a safe boundary after successful Undo Recovery, even for an idle bucket. Restoring it should not require the already compensated history.

Solution

Propagate acknowledged Undo write offsets back to the operator after all recovery writes complete. Preserve the scanned recovery boundary for buckets that need no compensating writes. This addresses stale baselines after successful recovery; it does not make arbitrary old checkpoints recoverable beyond log retention.

Are you willing to submit a PR?

• I'm willing to submit a PR!