Failed to restore committer state. This can happen when…

Error message

Failed to restore committer state. This can happen when operator uid changed and Flink allowNonRestoredState is enabled. Best practice is to explicitly set the operator id via FlinkSink#Builder#uidPrefix() so that the committer operator uid is stable. Otherwise, Flink auto generate an operator uid based on job topology.With that, operator uid is subjective to change upon topology change.

What it means

IcebergFilesCommitter.initializeState warns that no committer state (pending data files / job id) could be restored even though the operator is marked as restored. This typically happens when the operator uid changed and the restore ran with allowNonRestoredState, so the state was silently dropped; committed data may have been written but never committed to Iceberg.

Solutions

Set an explicit uidPrefix via FlinkSink.Builder#uidPrefix so the committer operator uid is stable across topology changes.
Restore from a savepoint/checkpoint taken with the same operator uid; avoid --allowNonRestoredState for production restores.
If data files were written but uncommitted, run removeOrphanFiles to clean them and restart with correct uids from a fresh snapshot.

Example fix

// before
FlinkSink.forRowData(input).tableLoader(loader).append();
// after
FlinkSink.forRowData(input).tableLoader(loader).uidPrefix("my-iceberg-sink").append();

Defensive patterns

Strategy: validation

Validate before calling

// verify the savepoint contains committer state before resuming
// and always set a stable operator uid
FlinkSink.forRowData(input).uidPrefix("iceberg-committer")...

Prevention

Always set uidPrefix on the sink builder
Never use --allowNonRestoredState for production restores
After topology changes, take a new savepoint before resuming
Check savepoint metadata for the committer operator uid before restart

When it happens

Trigger: Resuming from a savepoint/checkpoint where the committer operator's uid differs (topology changed, uidPrefix not set) while --allowNonRestoredState is enabled, causing jobIdState.getListState to return an empty/null iterable.

Common situations: Changing FlinkSink topology (adding/removing operators, parallelism changes with different uids) and restarting from an old savepoint; upgrading the job without a stable uidPrefix; careless savepoint with -n/--allowNonRestoredState.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/d0331f7a60e17824. Report an issue: GitHub.

Appendix: source

Thrown at flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:173

    maxContinuousEmptyCommits =
        PropertyUtil.propertyAsInt(table.properties(), MAX_CONTINUOUS_EMPTY_COMMITS, 10);
    Preconditions.checkArgument(
        maxContinuousEmptyCommits > 0, MAX_CONTINUOUS_EMPTY_COMMITS + " must be positive");

    int subTaskId = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
    int attemptId = getRuntimeContext().getTaskInfo().getAttemptNumber();
    this.manifestOutputFileFactory =
        FlinkManifestUtil.createOutputFileFactory(
            () -> table, table.properties(), flinkJobId, operatorUniqueId, subTaskId, attemptId);
    this.maxCommittedCheckpointId = INITIAL_CHECKPOINT_ID;

    this.checkpointsState = context.getOperatorStateStore().getListState(STATE_DESCRIPTOR);
    this.jobIdState = context.getOperatorStateStore().getListState(JOB_ID_DESCRIPTOR);
    if (context.isRestored()) {
      Iterable<String> jobIdIterable = jobIdState.get();
      if (jobIdIterable == null || !jobIdIterable.iterator().hasNext()) {
        LOG.warn(
            "Failed to restore committer state. This can happen when operator uid changed and Flink "
                + "allowNonRestoredState is enabled. Best practice is to explicitly set the operator id "
                + "via FlinkSink#Builder#uidPrefix() so that the committer operator uid is stable. "
                + "Otherwise, Flink auto generate an operator uid based on job topology."
                + "With that, operator uid is subjective to change upon topology change.");
        return;
      }

      String restoredFlinkJobId = jobIdIterable.iterator().next();
      Preconditions.checkState(
          !Strings.isNullOrEmpty(restoredFlinkJobId),
          "Flink job id parsed from checkpoint snapshot shouldn't be null or empty");

      // Since flink's checkpoint id will start from the max-committed-checkpoint-id + 1 in the new
      // flink job even if it's restored from a snapshot created by another different flink job, so
      // it's safe to assign the max committed checkpoint id from restored flink job to the current
      // flink job.
      this.maxCommittedCheckpointId =

View on GitHub (pinned to 86d9c8fc54)