[FLINK-40302][state] Do not let native savepoints block periodic materialization

[wonyongChoi05]

What is the purpose of the change

Fixes FLINK-40302: with the changelog state backend enabled, taking a single intermediate NATIVE-format savepoint permanently stops periodic materialization for the rest of the job's lifetime.

ChangelogKeyedStateBackend#nativeSavepoint() consumes a materialization ID (materializedId++) and registers it in materializationIdByCheckpointId, expecting a later notifyCheckpointComplete() to confirm it. However, the CheckpointCoordinator only sends acknowledge messages for savepoints that are synchronous (FLIP-203) — intermediate savepoints are never notified. Since the savepoint succeeded, lastFailedMaterializationId is not advanced either. The guard in initMaterialization()

lastConfirmedMaterializationId < materializedId - 1
    && lastFailedMaterializationId < materializedId - 1

therefore stays true forever, every subsequent periodic materialization is skipped ("materialization:{} not confirmed or failed or cancelled, skip trigger new one."), the delegated RocksDB backend is never flushed again, and the changelog grows unboundedly until the job is restarted.

The root cause is that the guard uses materializedId - 1 as a proxy for "the last triggered materialization", while nativeSavepoint() consumes IDs from the same counter without triggering a materialization, breaking the proxy.

Brief change log

• Introduce lastTriggeredMaterializationId in ChangelogKeyedStateBackend, set when initMaterialization() actually triggers a materialization and reset in completeRestore().
• The skip-guard in initMaterialization() now compares lastConfirmedMaterializationId / lastFailedMaterializationId against lastTriggeredMaterializationId instead of materializedId - 1.
• nativeSavepoint() is unchanged: it still consumes a unique ID and registers it for confirmation, so synchronous savepoints keep notifying the delegated backend.

Verifying this change

This change added tests and can be verified as follows:

Added ChangelogKeyedStateBackendTest#testInitMaterializationAfterAbortedNativeSavepoint : takes a NATIVE-format savepoint without delivering notifyCheckpointComplete (i.e. an intermediate savepoint), then asserts that initMaterialization() still triggers. Without this fix it returns Optional.empty(); with the fix it triggers as expected.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — the periodic materialization trigger path of the changelog state backend
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 79ae8d0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build