code: private void discardSubtaskState(
  			final JobID jobId,
  			final ExecutionAttemptID executionAttemptID,
  			final long checkpointId,
  			final TaskStateSnapshot subtaskState) {
  
  		if (subtaskState != null) {
  			executor.execute(new Runnable() {
  				@Override
  				public void run() {
  
  					try {
  						subtaskState.discardState();
  					} catch (Throwable t2) {
  						LOG.warn("Could not properly discard state object of checkpoint {} " +
  							"belonging to task {} of job {}.", checkpointId, executionAttemptID, jobId, t2);
  					}
  				}
  			});
  		}
  	}
docs: Discards the given state object asynchronously belonging to the given job, execution attempt
  id and checkpoint id.
  
  @param jobId identifying the job to which the state object belongs
  @param executionAttemptID identifying the task to which the state object belongs
  @param checkpointId of the state object
  @param subtaskState to discard asynchronously