Flink: Validate concurrent commits in IcebergSink to prevent commit duplication

[fmorillo7694]

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).