[FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Component/s: API / DataStream, Connectors / Common
• Labels: pull-request-available
• Assignee: Peter Vary
• Reporter: Peter Vary
• Created: 20/Oct/23 12:51
• Updated: 06/Feb/24 11:22
• Resolved: 06/Feb/24 11:22

Description

In case of the Iceberg Sink, we would like to use the WithPreCommitTopology to aggregate the writer results and create a single committable from them. So we would like to change both the type, and the number of the messages. Using the current WithPreCommitTopology interface we can work around the issue by using a Tuple, or POJO where some of the fields are used only before the addPreCommitTopology method, and some of the fields are only used after the method, but this seems more like abusing the interface than using it.

This is a more generic issue where the WithPreCommitTopology should provide a way to transform not only the data, but the type of the data channelled through it.

Issue Links

• duplicates: FLINK-33972 Enhance and synchronize Sink API to match the Source API
• links to: GitHub Pull Request #23587
• links to: GitHub Pull Request #23912