code: @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;
  			}
  		}
  	}
docs: ------------------------------------------------------------------------