apache/iceberg · warning

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 restores its pending checkpoints and flink jobId state in initializeState. If isRestored() is true but the jobId state is empty, the operator state could not be restored (changed operator uid with allowNonRestoredState) and the committer starts fresh, logging this detailed warning - previously pending files may be orphaned.

Solutions

1. Set an explicit stable uidPrefix via FlinkSink.Builder#uidPrefix(...) so the committer operator UID survives topology changes
2. Restore with allowNonRestoredState=false to fail fast instead of silently losing committer state
3. Run DeleteOrphanFiles afterward to clean files written by the lost pending commits
4. Re-run the job from the last successful checkpoint with the original topology

Example fix

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

// ensure a stable operator uid before appending the sink
Preconditions.checkState(uidPrefix != null, "Set FlinkSink.Builder#uidPrefix for stable committer state");
// and restore without dropping state
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(...);
// run restore with: flink run -s <savepoint> (WITHOUT --allowNonRestoredState)

When it happens

Trigger: initializeState during job restore where context.isRestored() but jobIdState.get() is null or empty: operator UID changed (topology change, missing uidPrefix) combined with Flink's allowNonRestoredState=true.

Common situations: Restarting after modifying the job graph without a stable uidPrefix; restoring a savepoint with allowNonRestoredState that silently dropped this operator's state; re-running the job from scratch against existing checkpoint files.

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

• Unknown serialize version:
• Unrecognized version or corrupt state
• Unrecognized version or corrupt state
• Can not alter the default database when the iceberg catalog…
• Cannot apply unknown modify-column change:

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

Appendix: source

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