code: public static CompletedCheckpointStore createCompletedCheckpoints(
  			CuratorFramework client,
  			Configuration configuration,
  			JobID jobId,
  			int maxNumberOfCheckpointsToRetain,
  			Executor executor) throws Exception {
  
  		checkNotNull(configuration, "Configuration");
  
  		String checkpointsPath = configuration.getString(
  			HighAvailabilityOptions.HA_ZOOKEEPER_CHECKPOINTS_PATH);
  
  		RetrievableStateStorageHelper<CompletedCheckpoint> stateStorage = createFileSystemStateStorage(
  			configuration,
  			"completedCheckpoint");
  
  		checkpointsPath += ZooKeeperSubmittedJobGraphStore.getPathForJob(jobId);
  
  		final ZooKeeperCompletedCheckpointStore zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore(
  			maxNumberOfCheckpointsToRetain,
  			createZooKeeperStateHandleStore(client, checkpointsPath, stateStorage),
  			executor);
  
  		LOG.info("Initialized {} in '{}'.", ZooKeeperCompletedCheckpointStore.class.getSimpleName(), checkpointsPath);
  		return zooKeeperCompletedCheckpointStore;
  	}
docs: Creates a {@link ZooKeeperCompletedCheckpointStore} instance.
  
  @param client                         The {@link CuratorFramework} ZooKeeper client to use
  @param configuration                  {@link Configuration} object
  @param jobId                          ID of job to create the instance for
  @param maxNumberOfCheckpointsToRetain The maximum number of checkpoints to retain
  @param executor to run ZooKeeper callbacks
  @return {@link ZooKeeperCompletedCheckpointStore} instance
  @throws Exception if the completed checkpoint store cannot be created