code: public void shutdown(JobStatus jobStatus) throws Exception {
  		synchronized (lock) {
  			if (!shutdown) {
  				shutdown = true;
  				LOG.info("Stopping checkpoint coordinator for job {}.", job);
  
  				periodicScheduling = false;
  				triggerRequestQueued = false;
  
  				// shut down the hooks
  				MasterHooks.close(masterHooks.values(), LOG);
  				masterHooks.clear();
  
  				// shut down the thread that handles the timeouts and pending triggers
  				timer.shutdownNow();
  
  				// clear and discard all pending checkpoints
  				for (PendingCheckpoint pending : pendingCheckpoints.values()) {
  					pending.abort(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
  				}
  				pendingCheckpoints.clear();
  
  				completedCheckpointStore.shutdown(jobStatus);
  				checkpointIdCounter.shutdown(jobStatus);
  			}
  		}
  	}
docs: Shuts down the checkpoint coordinator.
  
  <p>After this method has been called, the coordinator does not accept
  and further messages and cannot trigger any further checkpoints.