Apache Iceberg version

1.10.1 (latest release)

Query engine

None

Please describe the bug 🐞

Flink: DynamicIcebergSink operator UID is non-deterministic across JVM restarts, breaking last-state / savepoint recovery

Summary

org.apache.iceberg.flink.sink.dynamic.DynamicIcebergSink embeds a per-construction UUID.randomUUID() (sinkId) into the explicit .uid(…) of the pre-commit-topology operator. Every JVM that constructs the sink produces a different operator UID for that operator, which means the OperatorID hash Flink persists in checkpoints/savepoints never matches the operator found in the restored JobGraph. Under production conditions this breaks upgradeMode: last-state on the Flink Kubernetes Operator.

Environment

• iceberg-flink-runtime 1.10.0 and 1.10.1 (both released) — both affected.
• Flink 2.1.1
• Flink Kubernetes Operator with spec.job.upgradeMode: last-state
• Flink config relevant to the failure mode:

  execution.checkpointing.mode: EXACTLY_ONCE
  execution.checkpointing.unaligned.enabled: true
  parallelism.default: 3
  

The source of non-determinism

In DynamicIcebergSink.java, the constructor assigns sinkId = UUID.randomUUID().toString(), and addPreCommitTopology(...) sets the aggregator's uid as:

.uid(prefixIfNotNull(uidPrefix, sinkId + "-pre-commit-topology"));

All other sink operators (generator, updater, writer, committer) use deterministic uid suffixes via prefixIfNotNull(uidPrefix, "-<name>"). The pre-commit aggregator is the only one that mixes in the random sinkId.

Across two constructions of the same topology (same uidPrefix), we observe:

test-7801605e-7ad8-428f-b94a-4a9d6a984b71-pre-commit-topology   # run 1
test-671358db-0517-456e-951b-b9afd0dbb904-pre-commit-topology   # run 2 (same JVM, new sink instance)

Different strings → different MD5 hashes → different OperatorIDs persisted in the savepoint.

How the failure surfaces

Observed in production

Full redeploy via Flink K8s Operator upgradeMode: last-state. JobManager log on the new cluster:

CheckpointCoordinator  - Starting job <new> from savepoint file:/…/chk-N
JobMaster              - Job <new> failed.
org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
Caused by: java.lang.IllegalStateException: There is no operator for the state <hash>
    at StateAssignmentOperation.checkStateMappingCompleteness(StateAssignmentOperation.java:779)

<hash> is the MD5 of the pre-commit-topology uid that was active when the savepoint was taken, which no longer exists in the restored topology because a new sinkId was drawn.

Why it's not always fatal

The pre-commit aggregator (DynamicWriteResultAggregator) declares no user-managed state and flushes its internal buffer in prepareSnapshotPreBarrier. Under aligned checkpoints with low/no in-flight data, its OperatorState in the savepoint is empty, and Flink's restore logic treats it as a no-op:

Checkpoints - Skipping empty savepoint state for operator <hash>

Under unaligned checkpoints, Flink captures each operator's input channel state (in-flight records that overtook the barrier) and keys it by the operator ID. With parallelism > 1 and real throughput, that channel state is rarely empty — and that's when the restore fails hard with the IllegalStateException above.

This means the severity scales with load: light-load test environments see the warn log, production-scale deployments under unaligned.enabled: true see the crash.

Reproduction

Minimal (uid determinism only)

// Build the same topology twice in one JVM, compare operator uids.
DynamicIcebergSink.<T>forInput(source)
    .generator(generator)
    .catalogLoader(catalog)
    .uidPrefix("test")
    .append();
Set<String> uids1 = streamEnv.getStreamGraph().getStreamNodes().stream()
    .map(StreamNode::getTransformationUID).filter(Objects::nonNull).collect(toSet());

// Same code, new env, new sink:
Set<String> uids2 = /* ... */;

// All uids match EXCEPT "test-<UUID>-pre-commit-topology" which differs.
assertThat(uids1).isEqualTo(uids2);  // FAILS on 1.10.x

End-to-end (savepoint)

1. Build the sink topology with uidPrefix("test") and writeParallelism(3); configure execution.checkpointing.unaligned.enabled: true and a persistent state.checkpoints.dir.
2. Produce data so at least one checkpoint finishes with the committer having written a snapshot (Committed rowDelta to table ... in the TM log).
3. flink stop --savepointPath file:///tmp/sp <jid> — takes a savepoint, suspends the job.
4. Inspect the savepoint's _metadata for operator UID strings: you will see <uidPrefix>-<UUID-A>-pre-commit-topology.
5. flink run -d -s file:///tmp/sp/savepoint-… <job.jar> — new JVM, new sinkId.
6. JM log:

   Starting job <new> from savepoint …
   Skipping empty savepoint state for operator <hash-of-UUID-A uid>
   Restoring job <new> …
   

7. The same exercise on a production-parallelism + real-throughput job surfaces the IllegalStateException.

Proposed fix

Drop sinkId from the pre-commit-topology operator's .uid(...) so it uses the same deterministic-suffix pattern as the other operators:

         .transform(
-            prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"),
+            prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"),   // operator name can keep sinkId (cosmetic)
             typeInformation,
             new DynamicWriteResultAggregator(catalogLoader))
-        .uid(prefixIfNotNull(uidPrefix, sinkId + "-pre-commit-topology"));
+        .uid(prefixIfNotNull(uidPrefix, "-pre-commit-topology"));  // uid must be stable across JVMs

sinkId remains a per-instance random UUID because DynamicCommitter still needs a unique identifier for file separation across concurrent sinks writing the same table (the thread-pool name iceberg-committer-pool-<sinkId> is cosmetic but the file-naming invariant is real).

A corresponding test should assert operatorUIDs(build()) == operatorUIDs(build()).

Impact and severity

• Any Flink DynamicIcebergSink user on 1.10.x running with upgradeMode: last-state (or manual savepoint → restart) is affected.
• Users with unaligned.enabled: true and parallelism > 1 will hit the hard IllegalStateException on redeploy once traffic is high enough to leave in-flight records at barrier time.
• allowNonRestoredState: true is not a safe workaround: silently dropping channel state for the aggregator can diverge from the committer's pending-commit state and produce duplicate or missing commits.

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time