org.apache.flink.runtime.checkpoint

Class CheckpointCoordinator

java.lang.Object

org.apache.flink.runtime.checkpoint.CheckpointCoordinator

Direct Known Subclasses:

SavepointCoordinator

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.

Depending on the configured RecoveryMode, the behaviour of the CompletedCheckpointStore and CheckpointIDCounter change. The default standalone implementations don't support any recovery.

Field Summary

Fields

Modifier and Type	Field and Description
protected CheckpointIDCounter	checkpointIdCounter Checkpoint ID counter to ensure ascending IDs.
protected Object	lock Coordinator-wide lock to safeguard the checkpoint updates

Constructor Summary

Constructors

Constructor and Description
CheckpointCoordinator(JobID job, long baseInterval, long checkpointTimeout, ExecutionVertex[] tasksToTrigger, ExecutionVertex[] tasksToWaitFor, ExecutionVertex[] tasksToCommitTo, ClassLoader userClassLoader, CheckpointIDCounter checkpointIDCounter, CompletedCheckpointStore completedCheckpointStore, RecoveryMode recoveryMode)
CheckpointCoordinator(JobID job, long baseInterval, long checkpointTimeout, long minPauseBetweenCheckpoints, int maxConcurrentCheckpointAttempts, ExecutionVertex[] tasksToTrigger, ExecutionVertex[] tasksToWaitFor, ExecutionVertex[] tasksToCommitTo, ClassLoader userClassLoader, CheckpointIDCounter checkpointIDCounter, CompletedCheckpointStore completedCheckpointStore, RecoveryMode recoveryMode, CheckpointStatsTracker statsTracker)

Method Summary

Modifier and Type	Method and Description
ActorGateway	createActivatorDeactivator(akka.actor.ActorSystem actorSystem, UUID leaderSessionID)
protected long	getAndIncrementCheckpointId()
protected ActorGateway	getJobStatusListener()
int	getNumberOfPendingCheckpoints()
int	getNumberOfRetainedSuccessfulCheckpoints()
Map<Long,PendingCheckpoint>	getPendingCheckpoints()
List<CompletedCheckpoint>	getSuccessfulCheckpoints()
boolean	isShutdown()
protected void	onCancelCheckpoint(long canceledCheckpointId) Callback on cancellation of a checkpoint.
protected void	onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint) Callback on full acknowledgement of a checkpoint.
protected void	onShutdown() Callback on shutdown of the coordinator.
boolean	receiveAcknowledgeMessage(AcknowledgeCheckpoint message) Receives an AcknowledgeCheckpoint message and returns whether the message was associated with a pending checkpoint.
boolean	receiveDeclineMessage(DeclineCheckpoint message) Receives a DeclineCheckpoint message and returns whether the message was associated with a pending checkpoint.
boolean	restoreLatestCheckpointedState(Map<JobVertexID,ExecutionJobVertex> tasks, boolean errorIfNoCheckpoint, boolean allOrNothingState)
protected void	setJobStatusListener(ActorGateway jobStatusListener)
void	shutdown() Shuts down the checkpoint coordinator.
void	startCheckpointScheduler()
void	stopCheckpointScheduler()
boolean	triggerCheckpoint(long timestamp) Triggers a new checkpoint and uses the given timestamp as the checkpoint timestamp.
boolean	triggerCheckpoint(long timestamp, long nextCheckpointId) Triggers a new checkpoint and uses the given timestamp as the checkpoint timestamp.

Methods inherited from class java.lang.Object

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

Field Detail

lock

protected final Object lock

Coordinator-wide lock to safeguard the checkpoint updates

checkpointIdCounter

protected final CheckpointIDCounter checkpointIdCounter

Checkpoint ID counter to ensure ascending IDs. In case of job manager failures, these need to be ascending across job managers.

Constructor Detail

CheckpointCoordinator

public CheckpointCoordinator(JobID job,
                             long baseInterval,
                             long checkpointTimeout,
                             ExecutionVertex[] tasksToTrigger,
                             ExecutionVertex[] tasksToWaitFor,
                             ExecutionVertex[] tasksToCommitTo,
                             ClassLoader userClassLoader,
                             CheckpointIDCounter checkpointIDCounter,
                             CompletedCheckpointStore completedCheckpointStore,
                             RecoveryMode recoveryMode)
                      throws Exception

Throws:

Exception

CheckpointCoordinator

public CheckpointCoordinator(JobID job,
                             long baseInterval,
                             long checkpointTimeout,
                             long minPauseBetweenCheckpoints,
                             int maxConcurrentCheckpointAttempts,
                             ExecutionVertex[] tasksToTrigger,
                             ExecutionVertex[] tasksToWaitFor,
                             ExecutionVertex[] tasksToCommitTo,
                             ClassLoader userClassLoader,
                             CheckpointIDCounter checkpointIDCounter,
                             CompletedCheckpointStore completedCheckpointStore,
                             RecoveryMode recoveryMode,
                             CheckpointStatsTracker statsTracker)
                      throws Exception

Throws:

Exception

Method Detail

onShutdown

protected void onShutdown()

Callback on shutdown of the coordinator. Called in lock scope.

onCancelCheckpoint

protected void onCancelCheckpoint(long canceledCheckpointId)

Callback on cancellation of a checkpoint. Called in lock scope.

onFullyAcknowledgedCheckpoint

protected void onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint)

Callback on full acknowledgement of a checkpoint. Called in lock scope.

shutdown

public void shutdown()
              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()

triggerCheckpoint

public boolean triggerCheckpoint(long timestamp)
                          throws Exception

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

Parameters:

timestamp - The timestamp for the checkpoint.

Throws:

Exception

triggerCheckpoint

public boolean triggerCheckpoint(long timestamp,
                                 long nextCheckpointId)
                          throws Exception

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

Parameters:

timestamp - The timestamp for the checkpoint.

nextCheckpointId - The checkpoint ID to use for this checkpoint or -1 if the checkpoint ID counter should be queried.

Throws:

Exception

receiveDeclineMessage

public boolean receiveDeclineMessage(DeclineCheckpoint message)
                              throws Exception

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

Parameters:

message - Checkpoint decline from the task manager

Returns:

Flag indicating whether the declined checkpoint was associated with a pending checkpoint.

Throws:

Exception

receiveAcknowledgeMessage

public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message)
                                  throws Exception

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:

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

restoreLatestCheckpointedState

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

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

getAndIncrementCheckpointId

protected long getAndIncrementCheckpointId()

getJobStatusListener

protected ActorGateway getJobStatusListener()

setJobStatusListener

protected void setJobStatusListener(ActorGateway jobStatusListener)

startCheckpointScheduler

public void startCheckpointScheduler()

stopCheckpointScheduler

public void stopCheckpointScheduler()

createActivatorDeactivator

public ActorGateway createActivatorDeactivator(akka.actor.ActorSystem actorSystem,
                                               UUID leaderSessionID)