@ 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 ) ; }
Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones .