[flink] Route undo offsets by sink subtask

[platinumhamburg]

Purpose

Linked issue: close #3914 3914

Brief change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This PR fixes incorrect routing of Undo Recovery offset reports in the Flink sink when multiple sink subtasks run in the same JVM, by introducing subtask-scoped reporter keys and ownership-aware delegate unregistration to avoid cross-subtask (and delayed-close) interference.

Changes:

• Bind producer offset reporting to a (reporterGroupId, subtaskIndex) key and register/remove delegates with ownership checks.
• Plumb the runtime sink subtask index into sink writer creation so each writer creates a reporter for its own subtask.
• Add/extend unit tests covering routing correctness, delayed close behavior, and prevention of adding an Undo Recovery sink to multiple topologies.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperatorFactory.java	Introduces reporter group ID + subtask routing keys, a static delegate registry keyed by reporter key, and ownership-aware unregister.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperator.java	Derives reporter key from group ID + runtime subtask index; validates/fails fast on invalid offset reports; unregisters with ownership.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkSink.java	Passes subtask index into writer construction and creates subtask-bound reporters from the stored operator factory.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java	Extends the adapter’s writer creation hook to include the runtime subtask index.
fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java	Same subtask-index plumbing for Flink 1.18 adapter variant.
fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java	Same subtask-index plumbing for Flink 2.2 adapter variant (using TaskInfo).
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperatorFactoryTest.java	New tests asserting correct routing across subtasks, fail-fast behavior, and delayed-close safety.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlussSinkBuilderTest.java	Adds a regression test ensuring an Undo Recovery-enabled sink builder isn’t reused across multiple topologies.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[luoyuxia]

+1