[flink] Reduce heap retained by sink undo recovery #5119

Search before asking

• I searched existing issues and found no matching report.

Fluss version

main (development)

Please describe the bug

Sink undo recovery can retain substantial heap during large recoveries, even after asynchronous writes and individual buckets have completed.

UndoRecoveryExecutor accumulates every write-completion future until all buckets have been scanned. Each BucketRecoveryContext also retains its complete deduplication set until the entire recovery ends. In the filtered scanner path, selected records have their primary keys encoded once during filtering and again during submission.

With millions of distinct recovered keys, these retained objects and duplicate allocations can leave little space for decoding and write buffers, resulting in repeated full GC and slow recovery.

The issue can be reproduced by recovering many buckets, allowing some buckets to finish before the remaining ones, and inspecting retained futures, completed-bucket deduplication state, and primary-key encoding counts.

Solution

• Bound the futures retained for completion tracking, flushing buffered writes before waiting and preserving asynchronous failures and acknowledged offsets.
• Release a bucket's deduplication state after its final records have been processed and its fixed recovery range has been consumed. Keep recovery statistics independently.
• Reuse the encoded primary key of a selected record while preserving the distinction between per-poll reservation and submitted undo operations.

Keep the fixed recovery ranges, exact deduplication, first-valid-change semantics, and successful-write completion requirements intact.

Are you willing to submit a PR?

• I'm willing to submit a PR!