[FLINK-40339][table] Do not read buffer entries after removing them

[gustavodemorais]

What is the purpose of the change

WatermarkCompactingSinkMaterializer removed buffer entries while iterating and read their values afterwards. The value of a Map.Entry is undefined once the entry has been removed, and RocksDBMapEntry returns null for it, so consolidation on restore only works because the heap backend keeps removed values readable.

Brief change log

• Consume all buffered values in consolidateBufferToMinValue before mutating the buffer, and drop it with a single clear()
• Iterate values() instead of entries() in the ordered branch

Verifying this change

• WatermarkCompactingSinkMaterializerTest tests working
• Our only state backend that tests this code is heap which already worked accidently. Didn’t find a clean way to add additional testing without a adding a load of bloat but the current tests passing is a good coverage

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no - consolidation runs at most once per key per restore
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

2.1.220 (Claude Code) with Opus 5

[flinkbot]

CI report:

• 7de938b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[rkhachatryan]

This is simpler indeed, but RocksDB opens a new iterator here IIRC.
Opening an iterator is quite expensive, so this might cause a performance regression.

[gustavodemorais]

This is the branch for the unordered state backends - so Rocksdb is not affected by the change. Currently in open source, only heap backend uses this path. I don't have a strong opinion though, if you think the delete strategy is safer, I can go with that

[gustavodemorais]

Ops, that is not correct. I just went with the delete approach which means also less changes

[rkhachatryan]

LGTM

[geeknarrator]

LGTM. Should we add a test for this? I don't think it is covered by existing test, is it?

[gustavodemorais]

Hey @geeknarrator, our only state backend that tests this code is heap which already worked accidentally. Didn’t find a clean way to add additional testing without a adding a load of bloat but the current tests passing is a good coverage. If you have a good/clean suggestion on how to test this, let me know