rkhachatryan commented on a change in pull request #16575: URL: https://github.com/apache/flink/pull/16575#discussion_r678574282
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/track/TaskStateRegistry.java ########## @@ -0,0 +1,243 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.state.track; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.state.StateObject; + +import java.util.Collection; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.concurrent.Executor; + +/** + * A component responsible for managing state that is not shared across different TMs (i.e. not + * managed by JM). Managing state here means {@link StateObject#discardState() deletion} once not + * used. + * + * <p>Must not be shared between different jobs. + * + * <p>Explicitly aware of checkpointing so that tracking of associations between checkpoints and + * snapshots can be reused (instead of implementing in each backend). + * + * <h1>State sharing inside TM</h1> + * + * <p>Inside each TM, the registry <strong>must</strong> be shared across state backends on the same + * level as their state is, or higher. For example, if all state backends of operators of the same + * job in TM use the same file on DFS then they <strong>must</strong> share the same registry. OTH, + * if nothing is shared then the Registry can be per-backend (or shared). + * + * <h1>State sharing across TMs</h1> + * + * If a rescaling operation (or recovery with state sharing inside TM) results in some state being + * shared by multiple TMs (or TaskStateRegistries), such relevant shared states must be communicated + * to the registry so that it can ignore them. Implementations <strong>must</strong> ignore such + * shared states. + * + * <h1>State identity</h1> + * + * For the purpose of matching state objects from the different calls (and from previous runs), + * state can be identified by {@link StateObjectID}. A collection of such IDs can be obtained from + * the {@link StateObject} by recursively traversing it. This is implementation specific; however, + * it must be consistent across attempts, registries and backends. + * + * <h1>Basic usage</h1> + * + * <ol> + * <li>{@link #stateUsed(Set, Collection) register state usage} + * <li>{@link #checkpointStarting(String, long, boolean) start checkpoint 1} + * <li>{@link #stateNotUsed(String, StateObject) unregister state usage} + * <li>{@link #checkpointStarting(String, long, boolean) start checkpoint 2} (state is not used as + * it was unregistered) + * <li>{@link #checkpointSubsumed(String, long) subsume checkpoint 1} (state can be discarded) + * </ol> + * + * <p>It is not required to register state usage strictly before starting a checkpoint (mostly + * because it is usually not known which state will be included into the snapshot when the + * checkpoint starts). + * + * <p>Usually, state registration and un-registration is done automatically via {@link + * #snapshotTaken(String, StateObject, long, boolean)}. + * + * <h1>Consistency and dangling state</h1> + * + * There are several factors that can impact consistency: incremental checkpoints; out-of-order + * calls from a single backend; multiple backends; multiple concurrent checkpoints; savepoints; + * missing confirmations. Potential issues include: + * + * <ul> + * <li>Incremental checkpoint subsumption: if some state is shared across checkpoints then + * discarding an older checkpoint might invalidate subsequent checkpoints + * <li>Out-of-order checkpoint completion by a single backend: if a reported snapshot doesn't + * include some state anymore then that state can be discarded. Any <strong>earlier</strong> + * snapshots reported <strong>later</strong> still using this state later will be invalid + * <li>Out-of-order state (un)registration by multiple backends: if some state is shared then some + * backend might register and unregister its usage before the other starts using it. The + * other's backend snapshots with this state may be invalid. + * <li>Savepoint state deletion: savepoint should never be deleted unless aborted; even if there + * is no confirmation or it has be subsumed + * <li>With RETAIN_ON_CANCELLATION policy, last num-retained checkpoints shouldn't be discarded + * even if the job is cancelled (NOT IMPLEMENTED) + * </ul> + * + * To allow implementations to address the above issues, client MUST: + * + * <ul> + * <li>list all the backends {@link #stateUsed(Set, Collection) using some state} at once + * <li>list all state objects {@link #snapshotTaken(String, StateObject, long, boolean) used in a + * snapshot} at once + * <li>notify that the checkpoint is {@link #checkpointStarting(String, long, boolean) started} + * before {@link #stateNotUsed(String, StateObject) unregistering} any state this checkpoint + * might be using (if a state was unregistered after a checkpoint was started it can still be + * included into the snapshot) + * <li>in particular, report the {@link #snapshotTaken(String, StateObject, long, boolean)} + * snapshot} for a checkpoint only after notifying about its start + * <li>not {@link #checkpointStarting(String, long, boolean) start any checkpoint} that may have + * been {@link #checkpointSubsumed(String, long) subsumed} + * <li>for savepoints, call either {@link #snapshotTaken(String, StateObject, long, boolean) + * snapshotTaken} or {@link #checkpointAborted(String, long) aborted} (otherwise, state won't + * be deleted) + * <li>for checkpoints, call either {@link #checkpointSubsumed(String, long) subsumed} or {@link + * #checkpointAborted(String, long) aborted} (otherwise, state won't be deleted until {@link + * #jobTerminated(JobStatus)}) Review comment: The method was removed in 898fb0f1aa705f15d56e8a68f7a630c037b5af06, I'll update javadoc. -- 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]
