docs: @Override
  	public void run() {
  
  		// ----------------------------
  		//  Initial State transition
  		// ----------------------------
  		while (true) {
  			ExecutionState current = this.executionState;
  			if (current == ExecutionState.CREATED) {
  				if (transitionState(ExecutionState.CREATED, ExecutionState.DEPLOYING)) {
  					// success, we can start our work
  					break;
  				}
  			}
  			else if (current == ExecutionState.FAILED) {
  				// we were immediately failed. tell the TaskManager that we reached our final state
  				notifyFinalState();
  				if (metrics != null) {
  					metrics.close();
  				}
  				return;
  			}
  			else if (current == ExecutionState.CANCELING) {
  				if (transitionState(ExecutionState.CANCELING, ExecutionState.CANCELED)) {
  					// we were immediately canceled. tell the TaskManager that we reached our final state
  					notifyFinalState();
  					if (metrics != null) {
  						metrics.close();
  					}
  					return;
  				}
  			}
  			else {
  				if (metrics != null) {
  					metrics.close();
  				}
  				throw new IllegalStateException("Invalid state for beginning of operation of task " + this + '.');
  			}
  		}
  
  		// all resource acquisitions and registrations from here on
  		// need to be undone in the end
  		Map<String, Future<Path>> distributedCacheEntries = new HashMap<>();
  		AbstractInvokable invokable = null;
  
  		try {
  			// ----------------------------
  			//  Task Bootstrap - We periodically
  			//  check for canceling as a shortcut
  			// ----------------------------
  
  			// activate safety net for task thread
  			LOG.info("Creating FileSystem stream leak safety net for task {}", this);
  			FileSystemSafetyNet.initializeSafetyNetForThread();
  
  			blobService.getPermanentBlobService().registerJob(jobId);
  
  			// first of all, get a user-code classloader
  			// this may involve downloading the job's JAR files and/or classes
  			LOG.info("Loading JAR files for task {}.", this);
  
  			userCodeClassLoader = createUserCodeClassloader();
  			final ExecutionConfig executionConfig = serializedExecutionConfig.deserializeValue(userCodeClassLoader);
  
  			if (executionConfig.getTaskCancellationInterval() >= 0) {
  				// override task cancellation interval from Flink config if set in ExecutionConfig
  				taskCancellationInterval = executionConfig.getTaskCancellationInterval();
  			}
  
  			if (executionConfig.getTaskCancellationTimeout() >= 0) {
  				// override task cancellation timeout from Flink config if set in ExecutionConfig
  				taskCancellationTimeout = executionConfig.getTaskCancellationTimeout();
  			}
  
  			if (isCanceledOrFailed()) {
  				throw new CancelTaskException();
  			}
  
  			// ----------------------------------------------------------------
  			// register the task with the network stack
  			// this operation may fail if the system does not have enough
  			// memory to run the necessary data exchanges
  			// the registration must also strictly be undone
  			// ----------------------------------------------------------------
  
  			LOG.info("Registering task at network: {}.", this);
  
  			network.registerTask(this);
  
  			for (ResultPartition partition : producedPartitions) {
  				taskEventDispatcher.registerPartition(partition.getPartitionId());
  			}
  
  			// add metrics for buffers
  			this.metrics.getIOMetricGroup().initializeBufferMetrics(this);
  
  			// register detailed network metrics, if configured
  			if (taskManagerConfig.getConfiguration().getBoolean(TaskManagerOptions.NETWORK_DETAILED_METRICS)) {
  				// similar to MetricUtils.instantiateNetworkMetrics() but inside this IOMetricGroup
  				MetricGroup networkGroup = this.metrics.getIOMetricGroup().addGroup("Network");
  				MetricGroup outputGroup = networkGroup.addGroup("Output");
  				MetricGroup inputGroup = networkGroup.addGroup("Input");
  
  				// output metrics
  				for (int i = 0; i < producedPartitions.length; i++) {
  					ResultPartitionMetrics.registerQueueLengthMetrics(
  						outputGroup.addGroup(i), producedPartitions[i]);
  				}
  
  				for (int i = 0; i < inputGates.length; i++) {
  					InputGateMetrics.registerQueueLengthMetrics(
  						inputGroup.addGroup(i), inputGates[i]);
  				}
  			}
  
  			// next, kick off the background copying of files for the distributed cache
  			try {
  				for (Map.Entry<String, DistributedCache.DistributedCacheEntry> entry :
  						DistributedCache.readFileInfoFromConfig(jobConfiguration)) {
  					LOG.info("Obtaining local cache file for '{}'.", entry.getKey());
  					Future<Path> cp = fileCache.createTmpFile(entry.getKey(), entry.getValue(), jobId, executionId);
  					distributedCacheEntries.put(entry.getKey(), cp);
  				}
  			}
  			catch (Exception e) {
  				throw new Exception(
  					String.format("Exception while adding files to distributed cache of task %s (%s).", taskNameWithSubtask, executionId), e);
  			}
  
  			if (isCanceledOrFailed()) {
  				throw new CancelTaskException();
  			}
  
  			// ----------------------------------------------------------------
  			//  call the user code initialization methods
  			// ----------------------------------------------------------------
  
  			TaskKvStateRegistry kvStateRegistry = kvStateService.createKvStateTaskRegistry(jobId, getJobVertexId());
  
  			Environment env = new RuntimeEnvironment(
  				jobId,
  				vertexId,
  				executionId,
  				executionConfig,
  				taskInfo,
  				jobConfiguration,
  				taskConfiguration,
  				userCodeClassLoader,
  				memoryManager,
  				ioManager,
  				broadcastVariableManager,
  				taskStateManager,
  				aggregateManager,
  				accumulatorRegistry,
  				kvStateRegistry,
  				inputSplitProvider,
  				distributedCacheEntries,
  				producedPartitions,
  				inputGates,
  				taskEventDispatcher,
  				checkpointResponder,
  				taskManagerConfig,
  				metrics,
  				this);
  
  			// now load and instantiate the task's invokable code
  			invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass, env);
  
  			// ----------------------------------------------------------------
  			//  actual task core work
  			// ----------------------------------------------------------------
  
  			// we must make strictly sure that the invokable is accessible to the cancel() call
  			// by the time we switched to running.
  			this.invokable = invokable;
  
  			// switch to the RUNNING state, if that fails, we have been canceled/failed in the meantime
  			if (!transitionState(ExecutionState.DEPLOYING, ExecutionState.RUNNING)) {
  				throw new CancelTaskException();
  			}
  
  			// notify everyone that we switched to running
  			taskManagerActions.updateTaskExecutionState(new TaskExecutionState(jobId, executionId, ExecutionState.RUNNING));
  
  			// make sure the user code classloader is accessible thread-locally
  			executingThread.setContextClassLoader(userCodeClassLoader);
  
  			// run the invokable
  			invokable.invoke();
  
  			// make sure, we enter the catch block if the task leaves the invoke() method due
  			// to the fact that it has been canceled
  			if (isCanceledOrFailed()) {
  				throw new CancelTaskException();
  			}
  
  			// ----------------------------------------------------------------
  			//  finalization of a successful execution
  			// ----------------------------------------------------------------
  
  			// finish the produced partitions. if this fails, we consider the execution failed.
  			for (ResultPartition partition : producedPartitions) {
  				if (partition != null) {
  					partition.finish();
  				}
  			}
  
  			// try to mark the task as finished
  			// if that fails, the task was canceled/failed in the meantime
  			if (!transitionState(ExecutionState.RUNNING, ExecutionState.FINISHED)) {
  				throw new CancelTaskException();
  			}
  		}
  		catch (Throwable t) {
  
  			// unwrap wrapped exceptions to make stack traces more compact
  			if (t instanceof WrappingRuntimeException) {
  				t = ((WrappingRuntimeException) t).unwrap();
  			}
  
  			// ----------------------------------------------------------------
  			// the execution failed. either the invokable code properly failed, or
  			// an exception was thrown as a side effect of cancelling
  			// ----------------------------------------------------------------
  
  			try {
  				// check if the exception is unrecoverable
  				if (ExceptionUtils.isJvmFatalError(t) ||
  						(t instanceof OutOfMemoryError && taskManagerConfig.shouldExitJvmOnOutOfMemoryError())) {
  
  					// terminate the JVM immediately
  					// don't attempt a clean shutdown, because we cannot expect the clean shutdown to complete
  					try {
  						LOG.error("Encountered fatal error {} - terminating the JVM", t.getClass().getName(), t);
  					} finally {
  						Runtime.getRuntime().halt(-1);
  					}
  				}
  
  				// transition into our final state. we should be either in DEPLOYING, RUNNING, CANCELING, or FAILED
  				// loop for multiple retries during concurrent state changes via calls to cancel() or
  				// to failExternally()
  				while (true) {
  					ExecutionState current = this.executionState;
  
  					if (current == ExecutionState.RUNNING || current == ExecutionState.DEPLOYING) {
  						if (t instanceof CancelTaskException) {
  							if (transitionState(current, ExecutionState.CANCELED)) {
  								cancelInvokable(invokable);
  								break;
  							}
  						}
  						else {
  							if (transitionState(current, ExecutionState.FAILED, t)) {
  								// proper failure of the task. record the exception as the root cause
  								failureCause = t;
  								cancelInvokable(invokable);
  
  								break;
  							}
  						}
  					}
  					else if (current == ExecutionState.CANCELING) {
  						if (transitionState(current, ExecutionState.CANCELED)) {
  							break;
  						}
  					}
  					else if (current == ExecutionState.FAILED) {
  						// in state failed already, no transition necessary any more
  						break;
  					}
  					// unexpected state, go to failed
  					else if (transitionState(current, ExecutionState.FAILED, t)) {
  						LOG.error("Unexpected state in task {} ({}) during an exception: {}.", taskNameWithSubtask, executionId, current);
  						break;
  					}
  					// else fall through the loop and
  				}
  			}
  			catch (Throwable tt) {
  				String message = String.format("FATAL - exception in exception handler of task %s (%s).", taskNameWithSubtask, executionId);
  				LOG.error(message, tt);
  				notifyFatalError(message, tt);
  			}
  		}
  		finally {
  			try {
  				LOG.info("Freeing task resources for {} ({}).", taskNameWithSubtask, executionId);
  
  				// clear the reference to the invokable. this helps guard against holding references
  				// to the invokable and its structures in cases where this Task object is still referenced
  				this.invokable = null;
  
  				// stop the async dispatcher.
  				// copy dispatcher reference to stack, against concurrent release
  				final BlockingCallMonitoringThreadPool dispatcher = this.asyncCallDispatcher;
  				if (dispatcher != null && !dispatcher.isShutdown()) {
  					dispatcher.shutdownNow();
  				}
  
  				// free the network resources
  				releaseNetworkResources();
  
  				// free memory resources
  				if (invokable != null) {
  					memoryManager.releaseAll(invokable);
  				}
  
  				// remove all of the tasks library resources
  				libraryCache.unregisterTask(jobId, executionId);
  				fileCache.releaseJob(jobId, executionId);
  				blobService.getPermanentBlobService().releaseJob(jobId);
  
  				// close and de-activate safety net for task thread
  				LOG.info("Ensuring all FileSystem streams are closed for task {}", this);
  				FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
  
  				notifyFinalState();
  			}
  			catch (Throwable t) {
  				// an error in the resource cleanup is fatal
  				String message = String.format("FATAL - exception in resource cleanup of task %s (%s).", taskNameWithSubtask, executionId);
  				LOG.error(message, t);
  				notifyFatalError(message, t);
  			}
  
  			// un-register the metrics at the end so that the task may already be
  			// counted as finished when this happens
  			// errors here will only be logged
  			try {
  				metrics.close();
  			}
  			catch (Throwable t) {
  				LOG.error("Error during metrics de-registration of task {} ({}).", taskNameWithSubtask, executionId, t);
  			}
  		}
  	}