[FLINK-40584][state/rocksdb] Bound MapState and timer prefix iterators

[Savonitar]

What is the purpose of the change

RocksDB MapState and timer iterators currently check their key prefix in Java after a native seek returns. When state contains many tombstones, RocksDB can scan through unrelated key ranges while searching for the next live entry. This increases CPU usage during state access and can significantly delay timer queue initialization after restore, which performs a seek for every assigned key group.
This change sets a native exclusive upper bound for each iterator, limiting it to the map’s key-group/key/namespace prefix or the timer queue’s key-group prefix. It prevents scanning unrelated tombstones while preserving the intended results. The change also guards exhausted MapState resume seeks before advancing and clears inherited seek restrictions on the private options copy so they cannot omit valid entries.

Brief change log

• Check iterator validity before advancing after a MapState cache-refill seek.
• Bound MapState iteration, isEmpty(), and clear() to the current key-group/key/namespace prefix, and timer iteration to its key-group prefix.
• Give each bounded iterator its own ReadOptions and upper-bound Slice, closing them together with the native iterator.
• Enable auto-prefix mode and clear inherited prefix and lower-bound restrictions on the private options copy.

Verifying this change

This change added tests and can be verified as follows:

• Added RocksDBMapStateTest coverage for exhausted resume seeks and operations next to tombstone-heavy neighboring maps.
• Extended KeyGroupPartitionedPriorityQueueWithRocksDBStoreTest to verify bounded seeks during timer queue initialization and cache refill.
• Added RocksDBPrefixIteratorTest coverage for prefix-end calculation, native iteration boundaries, prefix extractors, inherited read options, snapshots, and resource cleanup.
• Confirmed that the relevant regression tests fail when the bounds, validity guard, or read-option protections are removed.
• Verified locally with JDK 17: all 51 focused cases passed, and the full RocksDB module unit suite completed with 881 executed cases passing. Spotless and Checkstyle passed.

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): (yes RocksDB MapState access and timer queue operations)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes, timer queue initialization during state restoration, e.g. after recovery)
• The S3 file system connector: (no)

Documentation

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

---

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

• Yes (please specify the tool below)

Generated-by: Claude Fable 5.1

[Savonitar]

cc: @pnowojski

[flinkbot]

CI report:

• a4ccbc8 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[pnowojski]

I have previously reviewed internal version of this PR. We have verified it on our production to resolve the performance issue when there are a lot of uncompacted deletion markers in the SST files.