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

A warning logged when the IcebergFilesCommitter operator restores from a checkpoint/savepoint but finds no persisted job ID state. This usually means the operator UID changed between jobs and Flink's allowNonRestoredState allowed the restore to proceed without the committer's state, so in-flight files from before cannot be committed.

Solutions

Set an explicit stable operator uid via FlinkSink.Builder#uidPrefix("my-sink-uid") so the committer operator uid survives topology changes
Restore from a checkpoint taken before the topology change, or without allowNonRestoredState so mismatches fail loudly instead of losing state
After a UID mismatch, expect orphaned data files from previous pending commits; clean them via expire/cleanup or a manual orphan-file removal

Example fix

// before
FlinkSink.forRowData(input).append();
// after
FlinkSink.forRowData(input).uidPrefix("iceberg-committer-stable").append();

Defensive patterns

Strategy: validation

Validate before calling

// Ensure a stable uid before building the sink
Preconditions.checkNotNull(uidPrefix, "uidPrefix must be set for the Iceberg committer");

Prevention

Always call uidPrefix() with a stable value on FlinkSink/IcebergSink builders
Avoid allowNonRestoredState in production restores so state mismatches fail loudly
Keep job topology changes and restore points coordinated

When it happens

Trigger: Restoring a Flink job whose topology changed (altering the auto-generated operator UID) with execution.checkpointing.allowed-state-restoration-types or allowNonRestoredState enabled; ListState for IcebergCommitState/JOB_ID_DESCRIPTOR is null or empty on restore.

Common situations: Job topology refactors (added/removed sink operators, renamed builders), missing uidPrefix() on the sink builder, upgrading Iceberg versions changing operator chains, restoring across Flink versions.

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/0ff2e5c8a94a8e7c. Report an issue: GitHub.

Appendix: source

Thrown at flink/v2.3/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)