@Override
	public boolean triggerCheckpoint(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime) throws Exception {

		try {
			// No alignment if we inject a checkpoint
			CheckpointMetrics checkpointMetrics = new CheckpointMetrics()
					.setBytesBufferedInAlignment(0L)
					.setAlignmentDurationNanos(0L);

			return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime);
		}
		catch (Exception e) {
			// propagate exceptions only if the task is still in "running" state
			if (isRunning) {
				throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() +
					" for operator " + getName() + '.', e);
			} else {
				LOG.debug("Could not perform checkpoint {} for operator {} while the " +
					"invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e);
				return false;
			}
		}
	}
@ Override public boolean triggerCheckpoint ( CheckpointMetaData checkpointMetaData , CheckpointOptions checkpointOptions , boolean advanceToEndOfEventTime ) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics ( ) . setBytesBufferedInAlignment ( 0L ) . setAlignmentDurationNanos ( 0L ) ; return performCheckpoint ( checkpointMetaData , checkpointOptions , checkpointMetrics , advanceToEndOfEventTime ) ; } catch ( Exception e ) { if ( isRunning ) { throw new Exception ( "Could not perform checkpoint " + checkpointMetaData . getCheckpointId ( ) + " for operator " + getName ( ) + '.' , e ) ; } else { LOG . debug ( "Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running." , checkpointMetaData . getCheckpointId ( ) , getName ( ) , e ) ; return false ; } } }
------------------------------------------------------------------------