fmorillo7694 opened a new pull request, #17901: URL: https://github.com/apache/iceberg/pull/17901
## Problem The non-dynamic `IcebergSink` committer (`IcebergCommitter`) is exposed to the duplicate-commit race described in #14425, which was fixed for the `DynamicIcebergSink` in #14517 (+ Flink 2.0/2.1 port in #14637) but never ported to the non-dynamic path. `IcebergCommitter.commit()` reads `getMaxCommittedCheckpointId()` once, up front, and then commits without re-validation: 1. The committer sends a commit for checkpoint N to the catalog; the request is slow (e.g. the catalog commits asynchronously after the client gives up) and the committer fails/restarts before receiving confirmation. 2. On recovery, Flink redelivers the committable. The restarted committer reads `getMaxCommittedCheckpointId()` — the original commit has not landed yet, so it decides checkpoint N is uncommitted. 3. The original commit now lands. When the redelivered commit reaches `SnapshotProducer.apply()`, the refresh picks up the just-landed snapshot as the new parent, so the second commit applies **cleanly — no conflict, no retry exhaustion** — and the same checkpoint's data files are committed twice. We reproduced this deterministically (test included) and verified the resulting table has 2 snapshots carrying the same `flink.max-committed-checkpoint-id` for the same job/operator id, with doubled record counts. The behavior is identical on HadoopTables, AWS Glue, and Amazon S3 Tables catalogs — once the interleaving occurs, the second commit is structurally valid from the catalog's perspective, so only the committer can prevent it. ## Solution Port the `MaxCommittedCheckpointIdValidator` introduced in #14517 to `IcebergCommitter`: register a `SnapshotAncestryValidator` on the commit operation that re-checks `max-committed-checkpoint-id` against the base snapshot ancestry **inside the commit transaction**, and skip the commit (log + return) when the branch already contains changes for the staged checkpoint — mirroring `DynamicCommitter`'s behavior. `SinkUtil.INITIAL_CHECKPOINT_ID` is widened from `private` to package-private so the validator can reuse it. The validator is currently duplicated as a private inner class (same shape as in `DynamicCommitter`). Happy to extract a shared class for both committers if preferred — kept the diff minimal for review. ## Testing - New `TestIcebergCommitterDuplicateCommit` reproduces the race deterministically (a delegating `TableLoader`/`Table` intercepts `AppendFiles.commit()` to land a concurrent commit of the same committable between the committer's dedup read and its commit). Fails on the unpatched committer with `snapshots=2 / doubled records`; passes with this change (`snapshots=1`, duplicate skipped). - `TestIcebergCommitter`: 288 tests, 0 failures (no regressions). - `TestDynamicCommitter`: 12/12 (unchanged). - `spotlessApply` clean. If the approach is accepted I'll follow up with the ports to the other Flink versions (v1.20, v2.0, v2.2, v2.3), mirroring how #14517 → #14637 was staged. Relates to #14425 (fixes the non-dynamic `IcebergSink` exposure it describes). -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
