[jira] [Created] (FLINK-40837) SavepointWriter#changeOperatorIdentifier drops the operator coordinator state

["Spoorthi Basu (Jira)" <ji...@apache.org>]

Spoorthi Basu created FLINK-40837:
-------------------------------------

             Summary: SavepointWriter#changeOperatorIdentifier drops the operator coordinator state
                 Key: FLINK-40837
                 URL: https://issues.apache.org/jira/browse/FLINK-40837
             Project: Flink
          Issue Type: Bug
          Components: API / State Processor
    Affects Versions: 2.1.3, 1.20.5, 2.2.1, 2.3.0, 2.0.2, 1.17.0
            Reporter: Spoorthi Basu


When a savepoint is rewritten with {{SavepointWriter#changeOperatorIdentifier}}, the operator's subtask state is moved to the new identifier but its coordinator state is not. For a source built on the {{Source}} API this is the split enumerator state. A job restored from the rewritten savepoint starts its source with {{Source#createEnumerator}} instead of {{Source#restoreEnumerator}}, as if it were a new job, while the readers restore their splits as usual. The restore succeeds with {{allowNonRestoredState=false}}. With {{NumberSequenceSource}}, the restored job emits the whole sequence a second time.

The code is the same in every release since 1.17.0 (checked at each release tag) and on current master, where I reproduced it (1690c6bed94).

Cause, at master:
* {{SavepointWriter.CheckpointMetadataCheckpointMetadataMapFunction}} (SavepointWriter.java:390) applies the identifier changes by calling {{OperatorState#copyWithNewIDs}} (line 423).
* {{OperatorState#copyWithNewIDs}} (OperatorState.java:178) builds a new {{OperatorState}} and copies only the subtask states. {{coordinatorState}} is not copied. FLINK-40836 is the same omission in {{OperatorState#copyAndDiscardInFlightData}}.

For comparison, when a savepoint is restored into a job that no longer has the operator, {{Checkpoints#loadAndValidateCheckpoint}} rejects it if that operator has coordinator state, unless {{allowNonRestoredState}} is set (Checkpoints.java:266). The rewrite drops the same state without any such check.

The method was added as {{copyWithNewOperatorID}} by FLINK-29457 (1.17.0), which introduced {{changeOperatorIdentifier}}, and renamed to {{copyWithNewIDs}} by FLINK-36001 (2.0.0). Neither version copies the coordinator state.

*Reproduction*

MiniCluster, parallelism 1. All operators have explicit uids.
# Run {{env.fromSequence(0, 9_999).uid("source-v1")}} followed by a map that sleeps 1 ms per record, and stop it with a savepoint after about 2,000 records.
# Rewrite the savepoint twice with {{SavepointWriter.fromExistingSavepoint(env, path)}}: once unchanged, once with {{changeOperatorIdentifier(OperatorIdentifier.forUid("source-v1"), OperatorIdentifier.forUid("source-v2"))}}.
# Restore the unchanged rewrite into the job with uid {{source-v1}}, and the changed rewrite into the same job with uid {{source-v2}}, both with {{allowNonRestoredState=false}} and without the sleep. Let each finish and count the records.

*Result*

Coordinator state of the source operator in each savepoint:
||Savepoint||NumberSequenceSource||Test source, 5,000 enumerator entries||
|original (uid source-v1)|16 bytes|520,016 bytes|
|rewritten, uid unchanged|16 bytes|520,016 bytes|
|rewritten, uid source-v1 to source-v2|none|none|

The subtask state was present under the new operator ID in the changed rewrite.

Records emitted by the restored {{fromSequence(0, 9_999)}} job:
||Restored from||Records||
|unchanged rewrite (uid source-v1)|7,988 (2,012 to 9,999), each once|
|changed rewrite (uid source-v2)|17,988: the same 7,988 plus all 10,000 values again|

A second run gave 7,991 and 17,991, again exactly 10,000 extra records. With the test source, the changed rewrite led to {{createEnumerator}} instead of {{restoreEnumerator}}, as in FLINK-40836.

*Fix*

{{copyWithNewIDs}} should carry the coordinator state over to the new {{OperatorState}} along with the subtask states. Coordinators are matched by operator ID on restore, so the state then reaches the coordinator of the renamed operator.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)