flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
/**
	 * Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
	 *
	 * @param checkpoint Completed checkpoint to add.
	 */
Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
@param checkpoint Completed checkpoint to add.
Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones . @param checkpoint Completed checkpoint to add .
@Override
	public void addCheckpoint(final CompletedCheckpoint checkpoint) throws Exception {
		checkNotNull(checkpoint, "Checkpoint");
		final String path = checkpointIdToPath(checkpoint.getCheckpointID());
		checkpointsInZooKeeper.addAndLock(path, checkpoint);
		completedCheckpoints.addLast(checkpoint);
		while (completedCheckpoints.size() > maxNumberOfCheckpointsToRetain) {
			try {
				removeSubsumed(completedCheckpoints.removeFirst());
			} catch (Exception e) {
				LOG.warn("Failed to subsume the old checkpoint", e);
			}
		}
		LOG.debug("Added {} to {}.", checkpoint, path);
	}
@ Override public void addCheckpoint ( final CompletedCheckpoint checkpoint ) throws Exception { checkNotNull ( checkpoint , "Checkpoint" ) ; final String path = checkpointIdToPath ( checkpoint . getCheckpointID ( ) ) ; checkpointsInZooKeeper . addAndLock ( path , checkpoint ) ; completedCheckpoints . addLast ( checkpoint ) ; while ( completedCheckpoints . size ( ) > maxNumberOfCheckpointsToRetain ) { try { removeSubsumed ( completedCheckpoints . removeFirst ( ) ) ; } catch ( Exception e ) { LOG . warn ( "Failed to subsume the old checkpoint" , e ) ; } } LOG . debug ( "Added {} to {}." , checkpoint , path ) ; }
Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones .
// Now add the new one. If it fails, we don't want to loose existing data. // Everything worked, let's remove a previous checkpoint if necessary.