Hi Flink CDC dev, I have a question about a design choice of Iceberg CDC pipeline connector. From what I see flink-cdc has its own implementation of iceberg writer and not using flink iceberg connector implementation. May I ask what was the context and motivation behind this design choice? From what i see, the latest version of flink-iceberg has many features that can benefit CDC integration such as: • https://iceberg.apache.org/docs/1.10.2/flink-writes/ dynamic sink: multi tables write and schema evolution • https://iceberg.apache.org/docs/1.10.2/flink-maintenance/ inline table maintenance I feel like this can make the flink-cdc integration falling behind with flink iceberg new features, for example the table maintenance in flink-cdc iceberg is using an old api of table maintenance and flink-iceberg has an implementation with better performance. Do you think it makes sense to reuse the flink iceberg connector with flink-cdc? If so, I'd happy to make a testing POC for this

Hi, thanks for the great question! Why Flink CDC has its own Iceberg writer implementation: The Flink CDC pipeline framework requires all sink connectors to implement Flink's SinkV2 API (org.apache.flink.api.connector.sink2.Sink). When the Iceberg connector was initially designed and added to Flink CDC, the project was built on JDK 8. At that time, the Iceberg Flink connector did not yet have a SinkV2-based implementation — the SinkV2 IcebergSink was introduced later in Iceberg PR #10179(https://github.com/apache/iceberg/pull/10179) (first released in Iceberg 1.8.0). Since no compatible SinkV2 implementation was available from Iceberg, Flink CDC had to build its own. Current state and future direction: Flink CDC has since moved to JDK 11, and Iceberg now offers both a single-table IcebergSink (SinkV2) and a multi-table DynamicIcebergSink (introduced in PR #13304(https://github.com/apache/iceberg/pull/13304), first released in Iceberg 1.10.0) — both still compatible with JDK 11 (up to Iceberg 1.10.x). So in principle, it is now possible to explore reusing the Iceberg Flink connector's implementation directly. That said, there are non-trivial challenges to consider: 1. Data abstraction mismatch: Flink CDC operates on its own Event type (DataChangeEvent, SchemaChangeEvent, etc.), while Iceberg's DynamicIcebergSink expects DynamicRecord/DynamicRecordInternal. A conversion layer between these two data models would be needed. 2. Schema evolution strategy differences: Flink CDC handles schema evolution externally via SchemaOperator + MetadataApplier (supporting add/drop/rename columns and type changes), while Iceberg's DynamicIcebergSink manages schema evolution internally through DynamicTableUpdateOperator (supporting add columns and type widening, but not drop or rename). These two approaches would need to be reconciled. We think it's worth exploring this direction. A POC would help validate the feasibility and surface any hidden issues. We'd be happy to discuss next steps based on the POC results — feel free to go ahead with it!

thanks a lot for sharing the knowledge @kunni 🙌. I'll give it a try and let you know