Package org.apache.flink.streaming.api.environment

Class CheckpointConfig

    • Field Detail

      • DEFAULT_TIMEOUT

        @Deprecated
public static final long DEFAULT_TIMEOUT
        Deprecated.
        This field is no longer used. Please use ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT instead.
        The default timeout of a checkpoint attempt: 10 minutes.

      • DEFAULT_MIN_PAUSE_BETWEEN_CHECKPOINTS

        @Deprecated
public static final long DEFAULT_MIN_PAUSE_BETWEEN_CHECKPOINTS
        Deprecated.
        This field is no longer used. Please use ExecutionCheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS instead.
        The default minimum pause to be made between checkpoints: none.

      • DEFAULT_MAX_CONCURRENT_CHECKPOINTS

        @Deprecated
public static final int DEFAULT_MAX_CONCURRENT_CHECKPOINTS
        Deprecated.
        This field is no longer used. Please use ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS instead.
        The default limit of concurrently happening checkpoints: one.

      • UNDEFINED_TOLERABLE_CHECKPOINT_NUMBER

        @Deprecated
public static final int UNDEFINED_TOLERABLE_CHECKPOINT_NUMBER
        Deprecated.
        This field is no longer used.
        See Also:
        Constant Field Values

      • DEFAULT_CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA

        @Deprecated
public static final int DEFAULT_CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA
        Deprecated.
        This field is no longer used. Please use ExecutionCheckpointingOptions.CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA instead.
        Default id of checkpoint for which in-flight data should be ignored on recovery.

    • Constructor Detail

      • CheckpointConfig

        public CheckpointConfig​(CheckpointConfig checkpointConfig)
        Creates a deep copy of the provided CheckpointConfig.
        Parameters:
        checkpointConfig - the config to copy.

      • CheckpointConfig

        public CheckpointConfig()

      • CheckpointConfig

        @Internal
public CheckpointConfig​(org.apache.flink.configuration.Configuration configuration)

    • Method Detail

      • disableCheckpointing

        public void disableCheckpointing()
        Disables checkpointing.

      • isCheckpointingEnabled

        public boolean isCheckpointingEnabled()
        Checks whether checkpointing is enabled.
        Returns:
        True if checkpointing is enables, false otherwise.

      • getCheckpointingMode

        public CheckpointingMode getCheckpointingMode()
        Gets the checkpointing mode (exactly-once vs. at-least-once).
        Returns:
        The checkpointing mode.

      • setCheckpointingMode

        public void setCheckpointingMode​(CheckpointingMode checkpointingMode)
        Sets the checkpointing mode (exactly-once vs. at-least-once).
        Parameters:
        checkpointingMode - The checkpointing mode.

      • getCheckpointInterval

        public long getCheckpointInterval()
        Gets the interval in which checkpoints are periodically scheduled.

        This setting defines the base interval. Checkpoint triggering may be delayed by the settings getMaxConcurrentCheckpoints() and getMinPauseBetweenCheckpoints().

        Returns:
        The checkpoint interval, in milliseconds.

      • setCheckpointInterval

        public void setCheckpointInterval​(long checkpointInterval)
        Sets the interval in which checkpoints are periodically scheduled.

        This setting defines the base interval. Checkpoint triggering may be delayed by the settings setMaxConcurrentCheckpoints(int) and setMinPauseBetweenCheckpoints(long).

        Parameters:
        checkpointInterval - The checkpoint interval, in milliseconds.

      • getCheckpointIntervalDuringBacklog

        public long getCheckpointIntervalDuringBacklog()
        Gets the interval in which checkpoints are periodically scheduled during backlog.

        This setting defines the base interval. Checkpoint triggering may be delayed by the settings getMaxConcurrentCheckpoints() and getMinPauseBetweenCheckpoints().

        If not explicitly configured, checkpoint interval during backlog will be the same as that in normal situation(see getCheckpointInterval()). If the return value is zero, it means that checkpoints would be disabled during backlog.

        Returns:
        The checkpoint interval, in milliseconds.

      • setCheckpointIntervalDuringBacklog

        public void setCheckpointIntervalDuringBacklog​(long checkpointInterval)
        Sets the interval in which checkpoints are periodically scheduled during backlog.

        This setting defines the base interval. Checkpoint triggering may be delayed by the settings setMaxConcurrentCheckpoints(int) and setMinPauseBetweenCheckpoints(long).

        If not explicitly configured, checkpoint interval during backlog will be the same as that in normal situation(see setCheckpointInterval(long)). If configured to zero, checkpoints would be disabled during backlog.

        Parameters:
        checkpointInterval - The checkpoint interval, in milliseconds.

      • getCheckpointTimeout

        public long getCheckpointTimeout()
        Gets the maximum time that a checkpoint may take before being discarded.
        Returns:
        The checkpoint timeout, in milliseconds.

      • setCheckpointTimeout

        public void setCheckpointTimeout​(long checkpointTimeout)
        Sets the maximum time that a checkpoint may take before being discarded.
        Parameters:
        checkpointTimeout - The checkpoint timeout, in milliseconds.

      • getMinPauseBetweenCheckpoints

        public long getMinPauseBetweenCheckpoints()
        Gets the minimal pause between checkpointing attempts. This setting defines how soon the checkpoint coordinator may trigger another checkpoint after it becomes possible to trigger another checkpoint with respect to the maximum number of concurrent checkpoints (see getMaxConcurrentCheckpoints()).
        Returns:
        The minimal pause before the next checkpoint is triggered.

      • setMinPauseBetweenCheckpoints

        public void setMinPauseBetweenCheckpoints​(long minPauseBetweenCheckpoints)
        Sets the minimal pause between checkpointing attempts. This setting defines how soon the checkpoint coordinator may trigger another checkpoint after it becomes possible to trigger another checkpoint with respect to the maximum number of concurrent checkpoints (see setMaxConcurrentCheckpoints(int)).

        If the maximum number of concurrent checkpoints is set to one, this setting makes effectively sure that a minimum amount of time passes where no checkpoint is in progress at all.

        Parameters:
        minPauseBetweenCheckpoints - The minimal pause before the next checkpoint is triggered.

      • getMaxConcurrentCheckpoints

        public int getMaxConcurrentCheckpoints()
        Gets the maximum number of checkpoint attempts that may be in progress at the same time. If this value is n, then no checkpoints will be triggered while n checkpoint attempts are currently in flight. For the next checkpoint to be triggered, one checkpoint attempt would need to finish or expire.
        Returns:
        The maximum number of concurrent checkpoint attempts.

      • setMaxConcurrentCheckpoints

        public void setMaxConcurrentCheckpoints​(int maxConcurrentCheckpoints)
        Sets the maximum number of checkpoint attempts that may be in progress at the same time. If this value is n, then no checkpoints will be triggered while n checkpoint attempts are currently in flight. For the next checkpoint to be triggered, one checkpoint attempt would need to finish or expire.
        Parameters:
        maxConcurrentCheckpoints - The maximum number of concurrent checkpoint attempts.

      • isForceCheckpointing

        @Deprecated
@PublicEvolving
public boolean isForceCheckpointing()
        Deprecated.
        This will be removed once iterations properly participate in checkpointing.
        Checks whether checkpointing is forced, despite currently non-checkpointable iteration feedback.
        Returns:
        True, if checkpointing is forced, false otherwise.

      • setForceCheckpointing

        @Deprecated
@PublicEvolving
public void setForceCheckpointing​(boolean forceCheckpointing)
        Deprecated.
        This will be removed once iterations properly participate in checkpointing.
        Checks whether checkpointing is forced, despite currently non-checkpointable iteration feedback.
        Parameters:
        forceCheckpointing - The flag to force checkpointing.

      • isForceUnalignedCheckpoints

        @PublicEvolving
public boolean isForceUnalignedCheckpoints()
        Checks whether unaligned checkpoints are forced, despite iteration feedback.
        Returns:
        True, if unaligned checkpoints are forced, false otherwise.

      • setForceUnalignedCheckpoints

        @PublicEvolving
public void setForceUnalignedCheckpoints​(boolean forceUnalignedCheckpoints)
        Checks whether unaligned checkpoints are forced, despite currently non-checkpointable iteration feedback or custom partitioners.
        Parameters:
        forceUnalignedCheckpoints - The flag to force unaligned checkpoints.

      • isFailOnCheckpointingErrors

        @Deprecated
public boolean isFailOnCheckpointingErrors()
        Deprecated.
        Use getTolerableCheckpointFailureNumber().
        This determines the behaviour when meeting checkpoint errors. If this returns true, which is equivalent to get tolerableCheckpointFailureNumber as zero, job manager would fail the whole job once it received a decline checkpoint message. If this returns false, which is equivalent to get tolerableCheckpointFailureNumber as the maximum of integer (means unlimited), job manager would not fail the whole job no matter how many declined checkpoints it received.

      • setFailOnCheckpointingErrors

        @Deprecated
public void setFailOnCheckpointingErrors​(boolean failOnCheckpointingErrors)
        Deprecated.
        Use setTolerableCheckpointFailureNumber(int).
        Sets the expected behaviour for tasks in case that they encounter an error when checkpointing. If this is set as true, which is equivalent to set tolerableCheckpointFailureNumber as zero, job manager would fail the whole job once it received a decline checkpoint message. If this is set as false, which is equivalent to set tolerableCheckpointFailureNumber as the maximum of integer (means unlimited), job manager would not fail the whole job no matter how many declined checkpoints it received.

        setTolerableCheckpointFailureNumber(int) would always overrule this deprecated method if they have conflicts.

      • getTolerableCheckpointFailureNumber

        public int getTolerableCheckpointFailureNumber()
        Get the defined number of consecutive checkpoint failures that will be tolerated, before the whole job is failed over.

        If the ExecutionCheckpointingOptions.TOLERABLE_FAILURE_NUMBER has not been configured, this method would return 0 which means the checkpoint failure manager would not tolerate any declined checkpoint failure.

      • setTolerableCheckpointFailureNumber

        public void setTolerableCheckpointFailureNumber​(int tolerableCheckpointFailureNumber)
        This defines how many consecutive checkpoint failures will be tolerated, before the whole job is failed over. The default value is `0`, which means no checkpoint failures will be tolerated, and the job will fail on first reported checkpoint failure.

      • setExternalizedCheckpointCleanup

        @PublicEvolving
public void setExternalizedCheckpointCleanup​(CheckpointConfig.ExternalizedCheckpointCleanup cleanupMode)
        Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled automatically unless the mode is set to CheckpointConfig.ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS.

        Externalized checkpoints write their meta data out to persistent storage and are not automatically cleaned up when the owning job fails or is suspended (terminating with job status JobStatus.FAILED or JobStatus.SUSPENDED). In this case, you have to manually clean up the checkpoint state, both the meta data and actual program state.

        The CheckpointConfig.ExternalizedCheckpointCleanup mode defines how an externalized checkpoint should be cleaned up on job cancellation. If you choose to retain externalized checkpoints on cancellation you have to handle checkpoint clean-up manually when you cancel the job as well (terminating with job status JobStatus.CANCELED).

        The target directory for externalized checkpoints is configured via CheckpointingOptions.CHECKPOINTS_DIRECTORY.

        Parameters:
        cleanupMode - Externalized checkpoint clean-up behaviour.

      • enableExternalizedCheckpoints

        @PublicEvolving
@Deprecated
public void enableExternalizedCheckpoints​(CheckpointConfig.ExternalizedCheckpointCleanup cleanupMode)
        Deprecated.
        use setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup) instead.
        Sets the mode for externalized checkpoint clean-up. Externalized checkpoints will be enabled automatically unless the mode is set to CheckpointConfig.ExternalizedCheckpointCleanup.NO_EXTERNALIZED_CHECKPOINTS.

        Externalized checkpoints write their meta data out to persistent storage and are not automatically cleaned up when the owning job fails or is suspended (terminating with job status JobStatus.FAILED or JobStatus.SUSPENDED). In this case, you have to manually clean up the checkpoint state, both the meta data and actual program state.

        The CheckpointConfig.ExternalizedCheckpointCleanup mode defines how an externalized checkpoint should be cleaned up on job cancellation. If you choose to retain externalized checkpoints on cancellation you have to handle checkpoint clean-up manually when you cancel the job as well (terminating with job status JobStatus.CANCELED).

        The target directory for externalized checkpoints is configured via CheckpointingOptions.CHECKPOINTS_DIRECTORY.

        Parameters:
        cleanupMode - Externalized checkpoint clean-up behaviour.

      • isExternalizedCheckpointsEnabled

        @PublicEvolving
public boolean isExternalizedCheckpointsEnabled()
        Returns whether checkpoints should be persisted externally.
        Returns:
        true if checkpoints should be externalized.

      • enableUnalignedCheckpoints

        @PublicEvolving
public void enableUnalignedCheckpoints​(boolean enabled)
        Enables unaligned checkpoints, which greatly reduce checkpointing times under backpressure.

        Unaligned checkpoints contain data stored in buffers as part of the checkpoint state, which allows checkpoint barriers to overtake these buffers. Thus, the checkpoint duration becomes independent of the current throughput as checkpoint barriers are effectively not embedded into the stream of data anymore.

        Unaligned checkpoints can only be enabled if ExecutionCheckpointingOptions.CHECKPOINTING_MODE is CheckpointingMode.EXACTLY_ONCE.

        Parameters:
        enabled - Flag to indicate whether unaligned are enabled.

      • enableUnalignedCheckpoints

        @PublicEvolving
public void enableUnalignedCheckpoints()
        Enables unaligned checkpoints, which greatly reduce checkpointing times under backpressure.

        Unaligned checkpoints contain data stored in buffers as part of the checkpoint state, which allows checkpoint barriers to overtake these buffers. Thus, the checkpoint duration becomes independent of the current throughput as checkpoint barriers are effectively not embedded into the stream of data anymore.

        Unaligned checkpoints can only be enabled if ExecutionCheckpointingOptions.CHECKPOINTING_MODE is CheckpointingMode.EXACTLY_ONCE.

      • isUnalignedCheckpointsEnabled

        @PublicEvolving
public boolean isUnalignedCheckpointsEnabled()
        Returns whether unaligned checkpoints are enabled.
        Returns:
        true if unaligned checkpoints are enabled.

      • isApproximateLocalRecoveryEnabled

        @Experimental
public boolean isApproximateLocalRecoveryEnabled()
        Returns whether approximate local recovery is enabled.
        Returns:
        true if approximate local recovery is enabled.

      • enableApproximateLocalRecovery

        @Experimental
public void enableApproximateLocalRecovery​(boolean enabled)
        Enables the approximate local recovery mode.

        In this recovery mode, when a task fails, the entire downstream of the tasks (including the failed task) restart.

        Notice that 1. Approximate recovery may lead to data loss. The amount of data which leads the failed task from the state of the last completed checkpoint to the state when the task fails is lost. 2. In the next version, we will support restarting the set of failed set of tasks only. In this version, we only support downstream restarts when a task fails. 3. It is only an internal feature for now.

        Parameters:
        enabled - Flag to indicate whether approximate local recovery is enabled .

      • getExternalizedCheckpointCleanup

        @PublicEvolving
public CheckpointConfig.ExternalizedCheckpointCleanup getExternalizedCheckpointCleanup()
        Returns the cleanup behaviour for externalized checkpoints.
        Returns:
        The cleanup behaviour for externalized checkpoints or null if none is configured.

      • setCheckpointStorage

        @Deprecated
@PublicEvolving
public void setCheckpointStorage​(org.apache.flink.runtime.state.CheckpointStorage storage)
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided for configuring checkpoint storage like the following code snippet: 
        
 Configuration config = new Configuration();
 config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
 config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///flink/checkpoints");
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        For more details on using ConfigOption for checkpoint storage configuration, please refer to the Flink documentation: Checkpoints
        CheckpointStorage defines how StateBackend's checkpoint their state for fault tolerance in streaming applications. Various implementations store their checkpoints in different fashions and have different requirements and availability guarantees.

        For example, JobManagerCheckpointStorage stores checkpoints in the memory of the JobManager. It is lightweight and without additional dependencies but is not highly available and only supports small state sizes. This checkpoint storage policy is convenient for local testing and development.

        FileSystemCheckpointStorage stores checkpoints in a filesystem. For systems like HDFS, NFS Drives, S3, and GCS, this storage policy supports large state size, in the magnitude of many terabytes while providing a highly available foundation for stateful applications. This checkpoint storage policy is recommended for most production deployments.

        Parameters:
        storage - The checkpoint storage policy.

      • setCheckpointStorage

        @Deprecated
@PublicEvolving
public void setCheckpointStorage​(String checkpointDirectory)
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided for configuring checkpoint storage like the following code snippet: 
        
 Configuration config = new Configuration();
 config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
 config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///flink/checkpoints");
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        For more details on using ConfigOption for checkpoint storage configuration, please refer to the Flink documentation: Checkpoints
        Configures the application to write out checkpoint snapshots to the configured directory. See FileSystemCheckpointStorage for more details on checkpointing to a file system.
        Parameters:
        checkpointDirectory - The path to write checkpoint metadata to.
        See Also:
        setCheckpointStorage(CheckpointStorage)

      • setCheckpointStorage

        @Deprecated
@PublicEvolving
public void setCheckpointStorage​(URI checkpointDirectory)
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided for configuring checkpoint storage like the following code snippet: 
        
 Configuration config = new Configuration();
 config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
 config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///flink/checkpoints");
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        For more details on using ConfigOption for checkpoint storage configuration, please refer to the Flink documentation: Checkpoints
        Configures the application to write out checkpoint snapshots to the configured directory. See FileSystemCheckpointStorage for more details on checkpointing to a file system.
        Parameters:
        checkpointDirectory - The path to write checkpoint metadata to.
        See Also:
        setCheckpointStorage(CheckpointStorage)

      • setCheckpointStorage

        @Deprecated
@PublicEvolving
public void setCheckpointStorage​(org.apache.flink.core.fs.Path checkpointDirectory)
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided for configuring checkpoint storage like the following code snippet: 
        
 Configuration config = new Configuration();
 config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
 config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///flink/checkpoints");
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        For more details on using ConfigOption for checkpoint storage configuration, please refer to the Flink documentation: Checkpoints
        Configures the application to write out checkpoint snapshots to the configured directory. See FileSystemCheckpointStorage for more details on checkpointing to a file system.
        Parameters:
        checkpointDirectory - The path to write checkpoint metadata to.
        See Also:
        setCheckpointStorage(String)

      • getCheckpointStorage

        @Deprecated
@Nullable
@PublicEvolving
public org.apache.flink.runtime.state.CheckpointStorage getCheckpointStorage()
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to find which checkpoint storage is used by checkpoint storage ConfigOption. For more details on using ConfigOption for checkpoint storage configuration, please refer to the Flink documentation: Checkpoints
        Returns:
        The CheckpointStorage that has been configured for the job. Or null if none has been set.
        See Also:
        setCheckpointStorage(CheckpointStorage)

      • setCheckpointIdOfIgnoredInFlightData

        @PublicEvolving
public void setCheckpointIdOfIgnoredInFlightData​(long checkpointIdOfIgnoredInFlightData)
        Setup the checkpoint id for which the in-flight data will be ignored for all operators in case of the recovery from this checkpoint.
        Parameters:
        checkpointIdOfIgnoredInFlightData - Checkpoint id for which in-flight data should be ignored.
        See Also:
        setCheckpointIdOfIgnoredInFlightData(long)

      • getCheckpointIdOfIgnoredInFlightData

        @PublicEvolving
public long getCheckpointIdOfIgnoredInFlightData()
        Returns:
        Checkpoint id for which in-flight data should be ignored.
        See Also:
        setCheckpointIdOfIgnoredInFlightData(long)

      • configure

        public void configure​(org.apache.flink.configuration.ReadableConfig configuration)
        Sets all relevant options contained in the ReadableConfig such as e.g. ExecutionCheckpointingOptions.CHECKPOINTING_MODE.

        It will change the value of a setting only if a corresponding option was set in the configuration. If a key is not present, the current value of a field will remain untouched.

        Parameters:
        configuration - a configuration to read the values from

      • toConfiguration

        @Internal
public org.apache.flink.configuration.Configuration toConfiguration()
        Returns:
        A copy of internal configuration. Note it is missing all options that are stored as plain java fields in CheckpointConfig, for example storage.