flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
/**
	 * Receives a {@link DeclineCheckpoint} message for a pending checkpoint.
	 *
	 * @param message Checkpoint decline from the task manager
	 */
Receives a  DeclineCheckpoint message for a pending checkpoint.
@param message Checkpoint decline from the task manager
Receives a DeclineCheckpoint message for a pending checkpoint . @param message Checkpoint decline from the task manager
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) {
			if (shutdown) {
				return;
			}
			checkpoint = pendingCheckpoints.remove(checkpointId);
			if (checkpoint != null && !checkpoint.isDiscarded()) {
				LOG.info("Decline checkpoint {} by task {}.", checkpointId, message.getTaskExecutionId());
				discardCheckpoint(checkpoint, message.getReason());
			}
			else if (checkpoint != null) {
				throw new IllegalStateException(
						"Received message for discarded but non-removed checkpoint " + checkpointId);
			}
			else if (LOG.isDebugEnabled()) {
				if (recentPendingCheckpoints.contains(checkpointId)) {
					LOG.debug("Received another decline message for now expired checkpoint attempt {} : {}",
							checkpointId, reason);
				} else {
					LOG.debug("Received decline message for unknown (too old?) checkpoint attempt {} : {}",
							checkpointId, reason);
				}
			}
		}
	}
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 ) { if ( shutdown ) { return ; } checkpoint = pendingCheckpoints . remove ( checkpointId ) ; if ( checkpoint != null && ! checkpoint . isDiscarded ( ) ) { LOG . info ( "Decline checkpoint {} by task {}." , checkpointId , message . getTaskExecutionId ( ) ) ; discardCheckpoint ( checkpoint , message . getReason ( ) ) ; } else if ( checkpoint != null ) { throw new IllegalStateException ( "Received message for discarded but non-removed checkpoint " + checkpointId ) ; } else if ( LOG . isDebugEnabled ( ) ) { if ( recentPendingCheckpoints . contains ( checkpointId ) ) { LOG . debug ( "Received another decline message for now expired checkpoint attempt {} : {}" , checkpointId , reason ) ; } else { LOG . debug ( "Received decline message for unknown (too old?) checkpoint attempt {} : {}" , checkpointId , reason ) ; } } } }
Receives a {@link DeclineCheckpoint} message for a pending checkpoint.
Receives a { @link DeclineCheckpoint } message for a pending checkpoint .
// we need to check inside the lock for being shutdown as well, otherwise we // get races and invalid error log messages // this should not happen // message is for an unknown checkpoint, or comes too late (checkpoint disposed) // message is for an unknown checkpoint. might be so old that we don't even remember it any more