org.apache.flink.runtime.checkpoint

Class CheckpointCoordinator

java.lang.Object

org.apache.flink.runtime.checkpoint.CheckpointCoordinator

public class CheckpointCoordinator
extends Object

The checkpoint coordinator coordinates the distributed snapshots of operators and state. It triggers the checkpoint by sending the messages to the relevant tasks and collects the checkpoint acknowledgements. It also collects and maintains the overview of the state handles reported by the tasks that acknowledge the checkpoint.

Constructor Summary

Constructors

Constructor and Description

CheckpointCoordinator(JobID job, long baseInterval, long checkpointTimeout, long minPauseBetweenCheckpoints, int maxConcurrentCheckpointAttempts, CheckpointRetentionPolicy retentionPolicy, ExecutionVertex[] tasksToTrigger, ExecutionVertex[] tasksToWaitFor, ExecutionVertex[] tasksToCommitTo, CheckpointIDCounter checkpointIDCounter, CompletedCheckpointStore completedCheckpointStore, StateBackend checkpointStateBackend, Executor executor, SharedStateRegistryFactory sharedStateRegistryFactory)

Method Summary

Modifier and Type	Method and Description
boolean	addMasterHook(MasterTriggerRestoreHook<?> hook) Adds the given master hook to the checkpoint coordinator.
JobStatusListener	createActivatorDeactivator()
void	failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause) Fails all pending checkpoints which have not been acknowledged by the given execution attempt id.
CheckpointIDCounter	getCheckpointIdCounter()
CheckpointStorage	getCheckpointStorage()
CompletedCheckpointStore	getCheckpointStore()
long	getCheckpointTimeout()
int	getNumberOfPendingCheckpoints()
int	getNumberOfRegisteredMasterHooks() Gets the number of currently register master hooks.
int	getNumberOfRetainedSuccessfulCheckpoints()
Map<Long,PendingCheckpoint>	getPendingCheckpoints()
List<CompletedCheckpoint>	getSuccessfulCheckpoints()
boolean	isPeriodicCheckpointingConfigured() Returns whether periodic checkpointing has been configured.
boolean	isShutdown()
boolean	receiveAcknowledgeMessage(AcknowledgeCheckpoint message) Receives an AcknowledgeCheckpoint message and returns whether the message was associated with a pending checkpoint.
void	receiveDeclineMessage(DeclineCheckpoint message) Receives a DeclineCheckpoint message for a pending checkpoint.
boolean	restoreLatestCheckpointedState(Map<JobVertexID,ExecutionJobVertex> tasks, boolean errorIfNoCheckpoint, boolean allowNonRestoredState) Restores the latest checkpointed state.
boolean	restoreSavepoint(String savepointPointer, boolean allowNonRestored, Map<JobVertexID,ExecutionJobVertex> tasks, ClassLoader userClassLoader) Restore the state with given savepoint.
void	setCheckpointStatsTracker(CheckpointStatsTracker statsTracker) Sets the checkpoint stats tracker.
void	shutdown(JobStatus jobStatus) Shuts down the checkpoint coordinator.
void	startCheckpointScheduler()
void	stopCheckpointScheduler()
boolean	triggerCheckpoint(long timestamp, boolean isPeriodic) Triggers a new standard checkpoint and uses the given timestamp as the checkpoint timestamp.
CheckpointTriggerResult	triggerCheckpoint(long timestamp, CheckpointProperties props, String externalSavepointLocation, boolean isPeriodic)
CompletableFuture<CompletedCheckpoint>	triggerSavepoint(long timestamp, String targetLocation) Triggers a savepoint with the given savepoint directory as a target.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

CheckpointCoordinator

public CheckpointCoordinator(JobID job,
                             long baseInterval,
                             long checkpointTimeout,
                             long minPauseBetweenCheckpoints,
                             int maxConcurrentCheckpointAttempts,
                             CheckpointRetentionPolicy retentionPolicy,
                             ExecutionVertex[] tasksToTrigger,
                             ExecutionVertex[] tasksToWaitFor,
                             ExecutionVertex[] tasksToCommitTo,
                             CheckpointIDCounter checkpointIDCounter,
                             CompletedCheckpointStore completedCheckpointStore,
                             StateBackend checkpointStateBackend,
                             Executor executor,
                             SharedStateRegistryFactory sharedStateRegistryFactory)

Method Detail

addMasterHook

public boolean addMasterHook(MasterTriggerRestoreHook<?> hook)

Adds the given master hook to the checkpoint coordinator. This method does nothing, if the checkpoint coordinator already contained a hook with the same ID (as defined via MasterTriggerRestoreHook.getIdentifier()).

Parameters:

hook - The hook to add.

Returns:

True, if the hook was added, false if the checkpoint coordinator already contained a hook with the same ID.

getNumberOfRegisteredMasterHooks

public int getNumberOfRegisteredMasterHooks()

Gets the number of currently register master hooks.

setCheckpointStatsTracker

public void setCheckpointStatsTracker(@Nullable
                                      CheckpointStatsTracker statsTracker)

Sets the checkpoint stats tracker.

Parameters:

statsTracker - The checkpoint stats tracker.

shutdown

public void shutdown(JobStatus jobStatus)
              throws Exception

Shuts down the checkpoint coordinator.

After this method has been called, the coordinator does not accept and further messages and cannot trigger any further checkpoints.

Throws:

Exception

isShutdown

public boolean isShutdown()

triggerSavepoint

public CompletableFuture<CompletedCheckpoint> triggerSavepoint(long timestamp,
                                                               @Nullable
                                                               String targetLocation)

Triggers a savepoint with the given savepoint directory as a target.

Parameters:

timestamp - The timestamp for the savepoint.

targetLocation - Target location for the savepoint, optional. If null, the state backend's configured default will be used.

Returns:

A future to the completed checkpoint

Throws:

IllegalStateException - If no savepoint directory has been specified and no default savepoint directory has been configured

triggerCheckpoint

public boolean triggerCheckpoint(long timestamp,
                                 boolean isPeriodic)

Triggers a new standard checkpoint and uses the given timestamp as the checkpoint timestamp.

Parameters:

timestamp - The timestamp for the checkpoint.

isPeriodic - Flag indicating whether this triggered checkpoint is periodic. If this flag is true, but the periodic scheduler is disabled, the checkpoint will be declined.

Returns:

true if triggering the checkpoint succeeded.

triggerCheckpoint

@VisibleForTesting
public CheckpointTriggerResult triggerCheckpoint(long timestamp,
                                                                    CheckpointProperties props,
                                                                    @Nullable
                                                                    String externalSavepointLocation,
                                                                    boolean isPeriodic)

receiveDeclineMessage

public void receiveDeclineMessage(DeclineCheckpoint message)

Receives a DeclineCheckpoint message for a pending checkpoint.

Parameters:

message - Checkpoint decline from the task manager

receiveAcknowledgeMessage

public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message)
                                  throws CheckpointException

Receives an AcknowledgeCheckpoint message and returns whether the message was associated with a pending checkpoint.

Parameters:

message - Checkpoint ack from the task manager

Returns:

Flag indicating whether the ack'd checkpoint was associated with a pending checkpoint.

Throws:

CheckpointException - If the checkpoint cannot be added to the completed checkpoint store.

failUnacknowledgedPendingCheckpointsFor

public void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId,
                                                    Throwable cause)

Fails all pending checkpoints which have not been acknowledged by the given execution attempt id.

Parameters:

executionAttemptId - for which to discard unacknowledged pending checkpoints

cause - of the failure

restoreLatestCheckpointedState

public boolean restoreLatestCheckpointedState(Map<JobVertexID,ExecutionJobVertex> tasks,
                                              boolean errorIfNoCheckpoint,
                                              boolean allowNonRestoredState)
                                       throws Exception

Restores the latest checkpointed state.

Parameters:

tasks - Map of job vertices to restore. State for these vertices is restored via Execution.setInitialState(JobManagerTaskRestore).

errorIfNoCheckpoint - Fail if no completed checkpoint is available to restore from.

allowNonRestoredState - Allow checkpoint state that cannot be mapped to any job vertex in tasks.

Returns:

true if state was restored, false otherwise.

Throws:

IllegalStateException - If the CheckpointCoordinator is shut down.

IllegalStateException - If no completed checkpoint is available and the failIfNoCheckpoint flag has been set.

IllegalStateException - If the checkpoint contains state that cannot be mapped to any job vertex in tasks and the allowNonRestoredState flag has not been set.

IllegalStateException - If the max parallelism changed for an operator that restores state from this checkpoint.

IllegalStateException - If the parallelism changed for an operator that restores non-partitioned state from this checkpoint.

Exception

restoreSavepoint

public boolean restoreSavepoint(String savepointPointer,
                                boolean allowNonRestored,
                                Map<JobVertexID,ExecutionJobVertex> tasks,
                                ClassLoader userClassLoader)
                         throws Exception

Restore the state with given savepoint.

Parameters:

savepointPointer - The pointer to the savepoint.

allowNonRestored - True if allowing checkpoint state that cannot be mapped to any job vertex in tasks.

tasks - Map of job vertices to restore. State for these vertices is restored via Execution.setInitialState(JobManagerTaskRestore).

userClassLoader - The class loader to resolve serialized classes in legacy savepoint versions.

Throws:

Exception

getNumberOfPendingCheckpoints

public int getNumberOfPendingCheckpoints()

getNumberOfRetainedSuccessfulCheckpoints

public int getNumberOfRetainedSuccessfulCheckpoints()

getPendingCheckpoints

public Map<Long,PendingCheckpoint> getPendingCheckpoints()

getSuccessfulCheckpoints

public List<CompletedCheckpoint> getSuccessfulCheckpoints()
                                                   throws Exception

Throws:

Exception

getCheckpointStorage

public CheckpointStorage getCheckpointStorage()

getCheckpointStore

public CompletedCheckpointStore getCheckpointStore()

getCheckpointIdCounter

public CheckpointIDCounter getCheckpointIdCounter()

getCheckpointTimeout

public long getCheckpointTimeout()

isPeriodicCheckpointingConfigured

public boolean isPeriodicCheckpointingConfigured()

Returns whether periodic checkpointing has been configured.

Returns:

true if periodic checkpoints have been configured.

startCheckpointScheduler

public void startCheckpointScheduler()

stopCheckpointScheduler

public void stopCheckpointScheduler()

createActivatorDeactivator

public JobStatusListener createActivatorDeactivator()