code: private void discardCheckpoint(PendingCheckpoint pendingCheckpoint, @Nullable Throwable cause) {
  		assert(Thread.holdsLock(lock));
  		Preconditions.checkNotNull(pendingCheckpoint);
  
  		final long checkpointId = pendingCheckpoint.getCheckpointId();
  
  		LOG.info("Discarding checkpoint {} of job {}.", checkpointId, job, cause);
  
  		if (cause == null || cause instanceof CheckpointDeclineException) {
  			pendingCheckpoint.abort(CheckpointFailureReason.CHECKPOINT_DECLINED, cause);
  		} else {
  			pendingCheckpoint.abort(CheckpointFailureReason.JOB_FAILURE, cause);
  		}
  
  		rememberRecentCheckpointId(checkpointId);
  
  		// we don't have to schedule another "dissolving" checkpoint any more because the
  		// cancellation barriers take care of breaking downstream alignments
  		// we only need to make sure that suspended queued requests are resumed
  
  		boolean haveMoreRecentPending = false;
  		for (PendingCheckpoint p : pendingCheckpoints.values()) {
  			if (!p.isDiscarded() && p.getCheckpointId() >= pendingCheckpoint.getCheckpointId()) {
  				haveMoreRecentPending = true;
  				break;
  			}
  		}
  
  		if (!haveMoreRecentPending) {
  			triggerQueuedRequests();
  		}
  	}
docs: Discards the given pending checkpoint because of the given cause.
  
  @param pendingCheckpoint to discard
  @param cause for discarding the checkpoint