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

### 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, "-")`. 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 from savepoint file:/…/chk-N
JobMaster - Job failed.
org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
Caused by: java.lang.IllegalStateException: There is no operator for the state
at StateAssignmentOperation.checkStateMappingCompleteness(StateAssignmentOperation.java:779)
```

`` 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
```

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.forInput(source)
.generator(generator)
.catalogLoader(catalog)
.uidPrefix("test")
.append();
Set uids1 = streamEnv.getStreamGraph().getStreamNodes().stream()
.map(StreamNode::getTransformationUID).filter(Objects::nonNull).collect(toSet());

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

// All uids match EXCEPT "test--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 ` — takes a savepoint, suspends the job.
4. Inspect the savepoint's `_metadata` for operator UID strings: you will see `--pre-commit-topology`.
5. `flink run -d -s file:///tmp/sp/savepoint-… ` — new JVM, new `sinkId`.
6. JM log:
```
Starting job from savepoint …
Skipping empty savepoint state for operator
Restoring job …
```
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-` 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

Start in DynamicIcebergSink.java, especially the constructor and addPreCommitTopology(...) where the pre-commit operator UID is assigned. Build the sink topology twice and compare the operator UIDs as described in the reproduction. Done means the pre-commit UID is stable across constructions while sinkId remains available for the committer, with a regression test covering the comparison.

Written by the indexing model from the issue text.