[FLINK-38539][table] Fix integer overflow in AbstractBytesMultiMap wh…

[nateab]

…en data exceeds 2GB

What is the purpose of the change

This pull request fixes an IndexOutOfBoundsException in AbstractBytesMultiMap that occurs when window aggregate queries (using TUMBLE TVF with LAST_VALUE aggregations) process large amounts of data with the
memory state backend. When the value/key area exceeds ~2GB (Integer.MAX_VALUE bytes), casting the long offset to int overflows to a negative value, causing invalid array access.

Brief change log

• Added bounds checking in writePointer() to validate offset before casting long to int
• Added bounds checking in skipPointer() to validate offset before casting long to int
• Added bounds checking in appendRecord() to validate key area offset before casting
• Removed redundant post-cast checks in appendValue() and appendRecord() since writePointer() now handles validation
• Ensures EOFException is thrown with a clear warning message before any overflow can occur

Verifying this change

This change added tests and can be verified as follows:

• Added testThrowsEOFExceptionWhenFull() test to BytesMultiMapTestBase that verifies EOFException is thrown when the map exceeds capacity (not IndexOutOfBoundsException which occurred before the fix)
• Existing tests (BytesMultiMapTest, WindowBytesMultiMapTest) continue to pass, verifying the fix doesn't break normal operation

The fix adds validation that throws EOFException earlier in the code path. The existing EOFException handling in RecordsWindowBuffer.addElement() already properly handles this case by flushing the buffer and
retrying.

##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 (only adds a comparison before existing cast)
• 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

[flinkbot]

CI report:

• 04e879b Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[davidradl]

Can we have a test with exactly the maximum, one under and one over (or as close as is feasible with the 4 byte pointer) to make sure that the boundary is exactly policed as expected please.

[nateab]

Sure I've added flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/util/collections/binary/AbstractBytesMultiMapBoundaryTest.java to test this now!