[flink] branch release-1.16 updated: [FLINK-29157][docs] Clarify the contract between CompletedCheckpointStore and SharedStateRegistry

[kl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

klion26 pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.16 by this push:
     new be2bd938385 [FLINK-29157][docs] Clarify the contract between CompletedCheckpointStore and SharedStateRegistry
be2bd938385 is described below

commit be2bd93838548f7858baecf5e8beb469836081d5
Author: fredia <fr...@gmail.com>
AuthorDate: Thu Oct 13 18:47:19 2022 +0800

    [FLINK-29157][docs] Clarify the contract between CompletedCheckpointStore and SharedStateRegistry
---
 .../flink/runtime/checkpoint/CompletedCheckpointStore.java | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
index df239a645a9..6abba9b03c4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint;
 
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.SharedStateRegistryFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,6 +40,16 @@ public interface CompletedCheckpointStore {
      * <p>Only a bounded number of checkpoints is kept. When exceeding the maximum number of
      * retained checkpoints, the oldest one will be discarded.
      *
+     * <p>After <a href="https://issues.apache.org/jira/browse/FLINK-24611">FLINK-24611</a>, {@link
+     * SharedStateRegistry#unregisterUnusedState} should be called here to subsume unused state.
+     * <font color="#FF0000"><strong>Note</strong></font>, the {@link CompletedCheckpoint} passed to
+     * {@link SharedStateRegistry#registerAllAfterRestored} or {@link
+     * SharedStateRegistryFactory#create} must be the same object as the input parameter, otherwise
+     * the state may be deleted by mistake.
+     *
+     * <p>After <a href="https://issues.apache.org/jira/browse/FLINK-25872">FLINK-25872</a>, {@link
+     * CheckpointsCleaner#cleanSubsumedCheckpoints} should be called explicitly here.
+     *
      * @return the subsumed oldest completed checkpoint if possible, return null if no checkpoint
      *     needs to be discarded on subsume.
      */
@@ -81,7 +92,8 @@ public interface CompletedCheckpointStore {
      * Shuts down the store.
      *
      * <p>The job status is forwarded and used to decide whether state should actually be discarded
-     * or kept.
+     * or kept. {@link SharedStateRegistry#unregisterUnusedState} and {@link
+     * CheckpointsCleaner#cleanSubsumedCheckpoints} should be called here to subsume unused state.
      *
      * @param jobStatus Job state on shut down
      * @param checkpointsCleaner that will cleanup completed checkpoints if needed