code: public void receiveDeclineMessage(DeclineCheckpoint message) {
  		if (shutdown || message == null) {
  			return;
  		}
  		if (!job.equals(message.getJob())) {
  			throw new IllegalArgumentException("Received DeclineCheckpoint message for job " +
  				message.getJob() + " while this coordinator handles job " + job);
  		}
  
  		final long checkpointId = message.getCheckpointId();
  		final String reason = (message.getReason() != null ? message.getReason().getMessage() : "");
  
  		PendingCheckpoint checkpoint;
  
  		synchronized (lock) {
  			// we need to check inside the lock for being shutdown as well, otherwise we
  			// get races and invalid error log messages
  			if (shutdown) {
  				return;
  			}
  
  			checkpoint = pendingCheckpoints.remove(checkpointId);
  
  			if (checkpoint != null && !checkpoint.isDiscarded()) {
  				LOG.info("Decline checkpoint {} by task {} of job {}.", checkpointId, message.getTaskExecutionId(), job);
  				discardCheckpoint(checkpoint, message.getReason());
  			}
  			else if (checkpoint != null) {
  				// this should not happen
  				throw new IllegalStateException(
  						"Received message for discarded but non-removed checkpoint " + checkpointId);
  			}
  			else if (LOG.isDebugEnabled()) {
  				if (recentPendingCheckpoints.contains(checkpointId)) {
  					// message is for an unknown checkpoint, or comes too late (checkpoint disposed)
  					LOG.debug("Received another decline message for now expired checkpoint attempt {} of job {} : {}",
  							checkpointId, job, reason);
  				} else {
  					// message is for an unknown checkpoint. might be so old that we don't even remember it any more
  					LOG.debug("Received decline message for unknown (too old?) checkpoint attempt {} of job {} : {}",
  							checkpointId, job, reason);
  				}
  			}
  		}
  	}
docs: Receives a {@link DeclineCheckpoint} message for a pending checkpoint.
  
  @param message Checkpoint decline from the task manager