@Override
    public void initializeState(StateInitializationContext context) throws Exception {
        super.initializeState(context);
        this.flinkJobId = getContainingTask().getEnvironment().getJobID().toString();
        this.operatorUniqueId = getRuntimeContext().getOperatorUniqueID();

        // Open the table loader and load the table.
        this.tableLoader.open();
        this.table = tableLoader.loadTable();
        this.committerMetrics = new IcebergFilesCommitterMetrics(super.metrics, table.name());

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

        int subTaskId = getRuntimeContext().getIndexOfThisSubtask();
        int attemptId = getRuntimeContext().getAttemptNumber();
        this.manifestOutputFileFactory =
                FlinkManifestUtil.createOutputFileFactory(
                        table, 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 =
                    getMaxCommittedCheckpointId(table, restoredFlinkJobId, operatorUniqueId, branch);

            NavigableMap<Long, byte[]> uncommittedDataFiles =
                    Maps.newTreeMap(checkpointsState.get().iterator().next())
                            .tailMap(maxCommittedCheckpointId, false);
            if (!uncommittedDataFiles.isEmpty()) {
                // Committed all uncommitted data files from the old flink job to iceberg table.
                long maxUncommittedCheckpointId = uncommittedDataFiles.lastKey();
                commitUpToCheckpoint(
                        uncommittedDataFiles, restoredFlinkJobId, operatorUniqueId, maxUncommittedCheckpointId);
            }
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergFilesCommitter.java