docs: public void triggerCheckpointBarrier(
  			final long checkpointID,
  			final long checkpointTimestamp,
  			final CheckpointOptions checkpointOptions,
  			final boolean advanceToEndOfEventTime) {
  
  		final AbstractInvokable invokable = this.invokable;
  		final CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointID, checkpointTimestamp);
  
  		if (executionState == ExecutionState.RUNNING && invokable != null) {
  
  			// build a local closure
  			final String taskName = taskNameWithSubtask;
  			final SafetyNetCloseableRegistry safetyNetCloseableRegistry =
  				FileSystemSafetyNet.getSafetyNetCloseableRegistryForThread();
  
  			Runnable runnable = new Runnable() {
  				@Override
  				public void run() {
  					// set safety net from the task's context for checkpointing thread
  					LOG.debug("Creating FileSystem stream leak safety net for {}", Thread.currentThread().getName());
  					FileSystemSafetyNet.setSafetyNetCloseableRegistryForThread(safetyNetCloseableRegistry);
  
  					try {
  						boolean success = invokable.triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime);
  						if (!success) {
  							checkpointResponder.declineCheckpoint(
  									getJobID(), getExecutionId(), checkpointID,
  									new CheckpointDeclineTaskNotReadyException(taskName));
  						}
  					}
  					catch (Throwable t) {
  						if (getExecutionState() == ExecutionState.RUNNING) {
  							failExternally(new Exception(
  								"Error while triggering checkpoint " + checkpointID + " for " +
  									taskNameWithSubtask, t));
  						} else {
  							LOG.debug("Encountered error while triggering checkpoint {} for " +
  								"{} ({}) while being not in state running.", checkpointID,
  								taskNameWithSubtask, executionId, t);
  						}
  					} finally {
  						FileSystemSafetyNet.setSafetyNetCloseableRegistryForThread(null);
  					}
  				}
  			};
  			executeAsyncCallRunnable(
  					runnable,
  					String.format("Checkpoint Trigger for %s (%s).", taskNameWithSubtask, executionId),
  					checkpointOptions.getCheckpointType().isSynchronous());
  		}
  		else {
  			LOG.debug("Declining checkpoint request for non-running task {} ({}).", taskNameWithSubtask, executionId);
  
  			// send back a message that we did not do the checkpoint
  			checkpointResponder.declineCheckpoint(jobId, executionId, checkpointID,
  					new CheckpointDeclineTaskNotReadyException(taskNameWithSubtask));
  		}
  	}