code: public void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause) {
  		synchronized (lock) {
  			Iterator<PendingCheckpoint> pendingCheckpointIterator = pendingCheckpoints.values().iterator();
  
  			while (pendingCheckpointIterator.hasNext()) {
  				final PendingCheckpoint pendingCheckpoint = pendingCheckpointIterator.next();
  
  				if (!pendingCheckpoint.isAcknowledgedBy(executionAttemptId)) {
  					pendingCheckpointIterator.remove();
  					discardCheckpoint(pendingCheckpoint, cause);
  				}
  			}
  		}
  	}
docs: Fails all pending checkpoints which have not been acknowledged by the given execution
  attempt id.
  
  @param executionAttemptId for which to discard unacknowledged pending checkpoints
  @param cause of the failure