Flink: Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart by lrsb · Pull Request #16011 · apache/iceberg

lrsb · 2026-04-17T12:40:21Z

The DynamicCommitter deduplicates commits on the triplet (flink.job-id, flink.operator-id, max-committed-checkpoint-id) stored in each snapshot summary. The aggregator previously sampled the job id from the runtime environment on every operator open(), so a restart or rescale with a fresh Flink JobID would stamp restored write results with the new id. The committer's ancestor walk then could not find the snapshot committed under the old id, and silently re-committed the data, producing duplicate rows in the target table.

Persist the aggregator's job id as operator state and reuse it on restore, so committables keep the job id under which their data was originally produced. This restores the dedup invariant across job restarts, autoscaler savepoint+resubmit cycles, and session-cluster resubmissions, which was broken under sustained catalog-side latency where commit responses were lost client-side.

pvary · 2026-04-20T11:53:06Z

                  writeToManifests(entries.getKey().tableName(), entries.getValue(), checkpointId),
-                  getContainingTask().getEnvironment().getJobID().toString(),
-                  getRuntimeContext().getOperatorUniqueID(),
+                  flinkJobId,


How did we handle this in the old sink? Would it be confusing if the jobId is not changing with the job?

I found similar logic in SinkV1 for deduplication based on job-id transformation, but I didn't see the related logic in SinkV2. Just like the dynamic sink, it directly obtains a new job-id.

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java

Lines 111 to 117 in bbd1212

// There're two cases that we restore from flink checkpoints: the first case is restoring from

// snapshot created by the same flink job; another case is restoring from snapshot created by

// another different job. For the second case, we need to maintain the old flink job's id in flink

// state backend to find the max-committed-checkpoint-id when traversing iceberg table's

// snapshots.

private static final ListStateDescriptor<String> JOB_ID_DESCRIPTOR =

new ListStateDescriptor<>("iceberg-flink-job-id", BasicTypeInfo.STRING_TYPE_INFO);

pvary · 2026-04-20T11:54:50Z

@lrsb: Let’s create the PR only for the latest Flink version. After we review and merge it, we can backport the changes to the other Flink versions. This makes it easier for the contributor (no need to keep multiple versions in sync) and for reviewers as well (fewer files to review).

pvary · 2026-04-20T11:55:05Z

CC: @mxm, @Guosmilesmile

mxm

[..] so a restart or rescale with a fresh Flink JobID would stamp restored write results with the new id.

Regular job restarts or rescalings do not change the Flink JobID. The JobID only changes on resubmission of the job, which can be triggered by the Kubernetes operator or some other external process.

That said, I'm not convinced we are fixing an actual bug here. DynamicCommitter will commit once the job restarts and restores its state. There will be no new entries in the commit state from the aggregator with the new JobID. The to-be-committed entries after restore will all carry the old JobID. So the logic will work fine as-is.

lrsb · 2026-04-21T12:50:15Z

[..] so a restart or rescale with a fresh Flink JobID would stamp restored write results with the new id.

Regular job restarts or rescalings do not change the Flink JobID. The JobID only changes on resubmission of the job, which can be triggered by the Kubernetes operator or some other external process.

That said, I'm not convinced we are fixing an actual bug here. DynamicCommitter will commit once the job restarts and restores its state. There will be no new entries in the commit state from the aggregator with the new JobID. The to-be-committed entries after restore will all carry the old JobID. So the logic will work fine as-is.

Regarding the job id change during rescalings, that happens when the scaling is applied using stop with savepoint and then the parallelism is overridden at vertex level, without involving the stock Flink autoscaler. I agree this is not the standard setup but might be a possibility when the autoscaler cannot be used.

mxm

Thanks @lrsb! The new solution looks good.

pvary · 2026-05-19T12:23:17Z

    assertThat(table.snapshots()).hasSize(2);

-    Snapshot snapshot1 = Iterables.getFirst(table.snapshots(), null);
-    assertThat(snapshot1.snapshotId()).isEqualTo(table.refs().get("branch1").snapshotId());


Why is this check removed?

Checking whether branch1 is always at index 0 of snapshots is wrong and it was making the test flaky

commitRequest1 and commitRequest2 creates branch1, and they are both for checkpointId1, only commitRequest3 should create branch2 which is checkpointId2.
How could it happen that branch2 is the first snapshot?
Can we check the history, that the commits are in order as we expect them?

How could it happen that branch2 is the first snapshot?

It can happen depending on the order of the commit request map iterable (basically the hashcode).

Can we check the history, that the commits are in order as we expect them?

I'd rather not add a dedicated history-order assertion as it would be testing the HashMap sort contract rather than user visible behavior, and testSkipsAlreadyCommittedDataAfterJobIdChanges already covers the dedup-correctness invariant under mixed jobIds

So the new code could result in cases where the commit from checkpoint 2 is added to the table sooner than a commit for checkpoint 1.

I can accept that this is ok, if the commits are for different branches. The question is: do we have tests that ensure that if we commit them in the same cycle, they are committed in the expected order?

The question is: do we have tests that ensure that if we commit them in the same cycle, they are committed in the expected order?

The order between different branches or different tables depends on the HashMap order, which can reorder a ckpt=2 branch-B commit before a ckpt=1 branch-A commit.

The question is: do we have tests that ensure that if we commit them in the same cycle, they are committed in the expected order?

The order of commits within the same cycle is currently not covered well, testSkipsAlreadyCommittedDataAfterJobIdChange covers just the single commit case. Will add a new test.

pvary · 2026-05-19T12:23:33Z

                .build());

-    Snapshot snapshot2 = Iterables.get(table.snapshots(), 1);
-    assertThat(snapshot2.snapshotId()).isEqualTo(table.refs().get("branch2").snapshotId());


Why is this check removed?

Same as above

stevenzwu · 2026-05-22T14:05:24Z

Is this problem only applicable to the dynamic sink?

DynamicCommitter dedupes pending commit requests against the table's ancestor snapshots by looking up snapshots whose FLINK_JOB_ID and OPERATOR_ID match the committable that produced them. The lookup used a single (jobId, operatorId) pair picked from the last committable in the batch, which breaks down when the batch spans multiple Flink jobIds. After a stop-with-savepoint, the committer's state replays in-flight committables tagged with the previous Flink jobId alongside fresh committables tagged with the new jobId. Using only the new jobId for dedup hides the snapshot that the previous jobId already produced, so the replayed committable's data files are committed a second time and the affected rows appear twice in the target table. Partition commit requests by (table, branch, jobId, operatorId) so each group is deduped and stamped with its own jobId. The downstream commit pipeline is unchanged; it already assumes a single (jobId, operatorId) per batch, which the new grouping restores in the mixed-jobId case. Change-Id: I2e0322a8f6564f960dd5c4730009c20e533223db

…y equals

lrsb · 2026-05-27T16:33:39Z

@stevenzwu The static IcebergFilesCommitter (in Sink V1) persists the previous flink.job-id in operator state and restores it on initializeState, so committables produced after a restart still carry the original jobId and the dedup walk finds the prior snapshot.

It's explicitly stated here

iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java

Lines 111 to 117 in bbd1212

    
           // There're two cases that we restore from flink checkpoints: the first case is restoring from 
        
           // snapshot created by the same flink job; another case is restoring from snapshot created by 
        
           // another different job. For the second case, we need to maintain the old flink job's id in flink 
        
           // state backend to find the max-committed-checkpoint-id when traversing iceberg table's 
        
           // snapshots. 
        
           private static final ListStateDescriptor<String> JOB_ID_DESCRIPTOR = 
        
               new ListStateDescriptor<>("iceberg-flink-job-id", BasicTypeInfo.STRING_TYPE_INFO);

pvary

+1 pending tests

lrsb · 2026-06-01T08:12:19Z

@pvary Seems I'm not allowed to merge, can you do it?

pvary · 2026-06-01T12:05:39Z

Merged to main.
Thanks @lrsb for the fix and all of the reviewers!

@lrsb: Please don't forget to create the backport PRs.
This could help:

git diff HEAD^ HEAD flink/v2,1 |sed "s/v2.1/v2.0/g">/tmp/patch;g apply -3 -p1 /tmp/patch

Backport commit apache#16011.

lrsb · 2026-06-01T15:20:24Z

@pvary backported in #16648

Backport commit #16011.

…nges on restart (apache#16011)

Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart (e.g. stop-with-savepoint, autoscaler resubmit). flink/v2.0's DynamicCommitter/TableKey/DynamicCommittable were byte-identical to flink/v2.1's pre-cherry-pick versions on this branch (both already on Iceberg 1.11.0), so this is a straight mirror of the v2.1 cherry-pick in the previous commit rather than a manual API adaptation.

…nges on restart (apache#16011) Cherry-pick of apache#16011. Upstream commit: apache/iceberg@c11404c

Backport of apache#16011 (apache/iceberg@c11404c) onto flink/v2.0. Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart (e.g. stop-with-savepoint, autoscaler resubmit). flink/v2.0's DynamicCommitter/TableKey/DynamicCommittable were byte-identical to flink/v2.1's pre-cherry-pick versions on this branch (both already on Iceberg 1.11.0), so this is a straight mirror of the v2.1 cherry-pick in the previous commit rather than a manual API adaptation.

Backport commit apache#16011.

…licate commits in DynamicCommitter when Flink jobId changes on restart (#10) * Flink: Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart (apache#16011) Cherry-pick of apache#16011. Upstream commit: apache/iceberg@c11404c * Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0 Backport commit apache#16011. * Pin version.txt to 1.11.0-affirm-patch.1 and add AFFIRM_VERSIONS.md version.txt was previously untracked, so build.gradle's getProjectVersion() silently falls back to a git-describe-derived version if the file is ever missing from the working tree (e.g. after a checkout/clean) instead of failing loudly. This has already caused one unintended publish under 1.10.0-SNAPSHOT. AFFIRM_VERSIONS.md records which Affirm-local patches are carried by each published version. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Remove published commit tracking from AFFIRM_VERSIONS.md Git tags are the source of truth for the commit published to Artifactory, not this file. * Document PR #11 (MinIO/Quay CI fix) in AFFIRM_VERSIONS.md Records the Docker Hub -> Quay MinIO fix as part of 1.11.0-affirm-patch.1 now that it's merged onto 1.11.0-affirm-patch. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: lrsb <github@lrsb.xyz> Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

github-actions Bot added the flink label Apr 17, 2026

lrsb marked this pull request as ready for review April 17, 2026 12:41

lrsb marked this pull request as draft April 17, 2026 13:15

lrsb marked this pull request as ready for review April 17, 2026 13:27

lrsb force-pushed the github branch from 6747d7d to fac1992 Compare April 17, 2026 13:30

pvary reviewed Apr 20, 2026

View reviewed changes

Comment thread .../flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java Outdated

pvary reviewed Apr 20, 2026

View reviewed changes

Guosmilesmile reviewed Apr 21, 2026

View reviewed changes

Comment thread .../flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java Outdated

lrsb marked this pull request as draft April 21, 2026 07:45

mxm reviewed Apr 21, 2026

View reviewed changes

Comment thread .../flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java

lrsb force-pushed the github branch from d3b004b to 6ca116d Compare April 21, 2026 12:30

lrsb changed the title ~~Flink: Persist flink job id in DynamicWriteResultAggregator state~~ Flink: Fix duplicate commits when Flink jobId changes on restart Apr 21, 2026

lrsb marked this pull request as ready for review April 21, 2026 12:33

mxm reviewed Apr 21, 2026

View reviewed changes

lrsb force-pushed the github branch 2 times, most recently from bb80d79 to 76b5f07 Compare April 21, 2026 15:56

aiborodin reviewed Apr 22, 2026

View reviewed changes

pvary reviewed Apr 22, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CommitGroupKey.java Outdated

pvary reviewed Apr 22, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CommitGroupKey.java Outdated

pvary reviewed Apr 22, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CommitGroupKey.java Outdated

pvary reviewed Apr 22, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java Outdated

aiborodin reviewed Apr 23, 2026

View reviewed changes

pvary reviewed Apr 27, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

pvary reviewed Apr 27, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java Outdated

pvary reviewed May 4, 2026

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

pvary reviewed May 19, 2026

View reviewed changes

lrsb added 5 commits May 27, 2026 18:14

Flink: Fix flaky testTableBranchAtomicCommitForAppendOnlyData

3953dae

Flink: Cache table ancestors per TableKey in DynamicCommitter

d891605

Flink: Order DynamicCommitter groups by checkpoint id

f5c70dc

Flink: Document DynamicCommitter grouping and null-safe JobOperatorKe…

d28ff9a

…y equals

lrsb force-pushed the github branch from e507dca to d28ff9a Compare May 27, 2026 16:36

stevenzwu changed the title ~~Flink: Fix duplicate commits when Flink jobId changes on restart~~ Flink: Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart May 28, 2026

Flink: Test DynamicCommitter commit order across jobIds

a3637c7

pvary approved these changes May 29, 2026

View reviewed changes

pvary merged commit c11404c into apache:main Jun 1, 2026
34 of 36 checks passed

lrsb added a commit to lrsb/iceberg that referenced this pull request Jun 1, 2026

Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0

96edcba

Backport commit apache#16011.

lrsb mentioned this pull request Jun 1, 2026

Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0 #16648

Merged

huaxingao pushed a commit that referenced this pull request Jun 1, 2026

Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0 (#16648)

7857238

Backport commit #16011.

nssalian added this to the Iceberg 1.12.0 milestone Jul 6, 2026

nickwu241 mentioned this pull request Sep 8, 2026

Flink: Cherry-pick #16011 and its v1.20/v2.0 backport - Fix duplicate commits in DynamicCommitter when Flink jobId changes on restart Affirm/iceberg#10

Merged

nickwu241 pushed a commit to Affirm/iceberg that referenced this pull request Sep 8, 2026

Flink: Fix duplicate commits in DynamicCommitter when Flink jobId cha…

8fb28c8

…nges on restart (apache#16011)

nickwu241 pushed a commit to Affirm/iceberg that referenced this pull request Sep 8, 2026

Flink: Fix duplicate commits in DynamicCommitter when Flink jobId cha…

50beb10

…nges on restart (apache#16011) Cherry-pick of apache#16011. Upstream commit: apache/iceberg@c11404c

nickwu241 pushed a commit to Affirm/iceberg that referenced this pull request Sep 8, 2026

Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0

ca5c888

Backport commit apache#16011.

nickwu241 pushed a commit to Affirm/iceberg that referenced this pull request Sep 8, 2026

Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0

be156de

Backport commit apache#16011.

jad-grepr mentioned this pull request Oct 5, 2026

Flink: Keep each writer's committed checkpoint in table properties #18374

Closed

rayhondo mentioned this pull request Oct 6, 2026

Flink: DynamicCommitter retry after an applied REST commit runs cleanAll() and deletes the committed snapshot's manifest list and manifests #18402

Open

1 task done

	// There're two cases that we restore from flink checkpoints: the first case is restoring from
	// snapshot created by the same flink job; another case is restoring from snapshot created by
	// another different job. For the second case, we need to maintain the old flink job's id in flink
	// state backend to find the max-committed-checkpoint-id when traversing iceberg table's
	// snapshots.
	private static final ListStateDescriptor<String> JOB_ID_DESCRIPTOR =
	new ListStateDescriptor<>("iceberg-flink-job-id", BasicTypeInfo.STRING_TYPE_INFO);

Conversation

lrsb commented Apr 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Guosmilesmile Apr 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pvary commented Apr 20, 2026

Uh oh!

pvary commented Apr 20, 2026

Uh oh!

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lrsb commented Apr 21, 2026

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu commented May 22, 2026

Uh oh!

lrsb commented May 27, 2026

Uh oh!

pvary left a comment

Choose a reason for hiding this comment

Uh oh!

lrsb commented Jun 1, 2026

Uh oh!

Uh oh!

pvary commented Jun 1, 2026

Uh oh!

lrsb commented Jun 1, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

lrsb commented Apr 17, 2026 •

edited

Loading

Guosmilesmile Apr 21, 2026 •

edited

Loading