inesmestre opened a new issue, #16128:
URL: https://github.com/apache/iceberg/issues/16128
### 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:
```java
.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 `OperatorID`s 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)
```java
// 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:
```diff
.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
--
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]
