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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
