code: @Override
  	public void addCheckpoint(final CompletedCheckpoint checkpoint) throws Exception {
  		checkNotNull(checkpoint, "Checkpoint");
  
  		final String path = checkpointIdToPath(checkpoint.getCheckpointID());
  
  		// Now add the new one. If it fails, we don't want to loose existing data.
  		checkpointsInZooKeeper.addAndLock(path, checkpoint);
  
  		completedCheckpoints.addLast(checkpoint);
  
  		// Everything worked, let's remove a previous checkpoint if necessary.
  		while (completedCheckpoints.size() > maxNumberOfCheckpointsToRetain) {
  			final CompletedCheckpoint completedCheckpoint = completedCheckpoints.removeFirst();
  			tryRemoveCompletedCheckpoint(completedCheckpoint, CompletedCheckpoint::discardOnSubsume);
  		}
  
  		LOG.debug("Added {} to {}.", checkpoint, path);
  	}
docs: Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
  
  @param checkpoint Completed checkpoint to add.