fcfangcc commented on PR #4360:
URL: https://github.com/apache/flink-cdc/pull/4360#issuecomment-4182423791
@spoorthibasu
The current fix uses a per-writer batchIndex counter to track flush
batches within a checkpoint. However, when the IcebergWriter runs
with parallelism > 1, each subtask maintains its own independent
tableBatchIndexMap. This causes a critical issue when multiple subtasks
receive the same SchemaChangeEvent for the same table.
Here is my implementation, You can refer to it
[flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java](https://github.com/fcfangcc/flink-cdc/blob/aa8b6fb48fb2d94c95f3fc9c348f055a66e34f09/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java)
* The eventId of the same SchemaChangeEvent is the same on all subtasks
* Data from different subtasks but belonging to the same schema change will
be merged into a snapshot
* LinkedHashMap guarantees to process according to the sequence of eventId
insertion (that is, chronological order)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
