[PR] [FLINK-40302][state] Fix materialization after aborted native savepoint [flink]

["wonyongChoi05 (via GitHub)" <gi...@apache.org>]

wonyongChoi05 opened a new pull request, #28891:
URL: https://github.com/apache/flink/pull/28891

   ## 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.
   This change tracks the last actually-triggered materialization ID and guards
   on that instead. Savepoint-consumed IDs no longer participate in the guard,
   while the original protection from #22669 (do not trigger a new
   materialization while the previous one is unresolved, to keep the
   SharedStateRegistry lineage continuous) is fully preserved — including the
   interleaving where a savepoint is taken while a materialization is still in
   flight.
   
   ## 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


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40302][state] Fix materialization after aborted native savepoint [flink]

["flinkbot (via GitHub)" <gi...@apache.org>]

flinkbot commented on PR #28891:
URL: https://github.com/apache/flink/pull/28891#issuecomment-5166185686

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "e2d99c506156818f9f9fea87da3d2c9458c0c9ca",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e2d99c506156818f9f9fea87da3d2c9458c0c9ca",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * e2d99c506156818f9f9fea87da3d2c9458c0c9ca UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40302][state] Do not let native savepoints block periodic materialization [flink]

["wonyongChoi05 (via GitHub)" <gi...@apache.org>]

wonyongChoi05 commented on PR #28891:
URL: https://github.com/apache/flink/pull/28891#issuecomment-5558198961

   Hi @rkhachatryan @masteryhx, could you take a look at this PR when you have a chance? I’m reaching out given your work on changelog materialization and native savepoints.
   This addresses FLINK-40302, where an intermediate native savepoint permanently blocks periodic materialization. A regression test is included, and Azure CI has passed. If someone else would be a better reviewer, could you point me to them? Thanks!****


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org