[flink] Fix $binlog update pairing across log splits

[luoyuxia]

Purpose

Linked issue: close #2879

When one Flink SourceReader processes records from multiple log splits, $binlog
can interleave UPDATE_BEFORE / UPDATE_AFTER records from different splits.
The previous implementation kept only one global pending UPDATE_BEFORE record,
which could be overwritten by another split and then fail with:

Received UPDATE_AFTER (+U) without a preceding UPDATE_BEFORE (-U) record.

Brief change log

• keep pending UPDATE_BEFORE state per split instead of using one global buffer;
• pass the current split id through the emitter/deserialization path for $binlog;
• pair UPDATE_BEFORE and UPDATE_AFTER only within the same split;
• add a regression test for cross-split interleaving in BinlogRowConverterTest.

Tests

• Added BinlogRowConverterTest#testCrossSplitInterleavingDoesNotCorrupt
• Not run locally

Also verify in production env

API and Format

No API or storage format changes.

Documentation

No.

[Copilot]

Pull request overview

This PR improves correctness of the Flink $binlog virtual table conversion by preventing UPDATE_BEFORE/UPDATE_AFTER pairing from being corrupted when records from multiple log splits are interleaved within the same source reader.

Changes:

• Add per-split buffering in BinlogRowConverter and a split-aware toBinlogRowData(record, splitId) overload.
• Propagate split context to BinlogDeserializationSchema via FlinkRecordEmitter (using a new SourceSplitState.splitId() accessor).
• Add a unit test covering cross-split interleaving behavior.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java	Adds a test ensuring interleaved splits don’t cross-pair update before/after images.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java	Switches update buffering from a single slot to a per-split map and adds a split-aware conversion API.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/SourceSplitState.java	Adds a splitId() accessor to avoid reconstructing split objects just to get the id.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java	Sets the current split id into the binlog deserializer before deserializing log records.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java	Tracks the current split id and routes records to the split-aware converter method.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.