A TaskRunner object is created when an executor is requested to launch a task.

It is created with an ExecutorBackend (to send the task’s status updates to), task and attempt ids, task name, and serialized version of the task (as ByteBuffer).

When run is executed, it creates a TaskMemoryManager object (using the global MemoryManager and the constructor’s taskId) to manage the memory allocated for the task’s execution.

It starts measuring the time to deserialize a task.

It sets the current context classloader.

It creates a new instance of the global closure Serializer.

You should see the following INFO message in the logs:

At this point, the task is considered running and the ExecutorBackend.statusUpdate is executed (with taskId and TaskState.RUNNING state).

run deserializes the task’s environment (from serializedTask bytes using Task.deserializeWithDependencies) to have the task’s files, jars and properties, and the bytes (i.e. the real task’s body).

updateDependencies(taskFiles, taskJars) is called.

This is the moment when the proper Task object is deserialized (from taskBytes) using the earlier-created closure Serializer object. The local properties (as localProperties) are initialized to be the task’s properties (from the earlier call to Task.deserializeWithDependencies) and the TaskMemoryManager (created earlier in the method) is set to the task.

If kill method has been called in the meantime, the execution stops by throwing a TaskKilledException. Otherwise, TaskRunner continues executing the task.

You should see the following DEBUG message in the logs:

TaskRunner sends update of the epoch of the task to MapOutputTracker.

The taskStart time which corresponds to the current time is recorded.

The task runs (with taskId, attemptNumber, and the globally-configured MetricsSystem). It runs inside a "monitored" block (i.e. try-finally block) to clean up after the task’s run finishes regardless of the final outcome - the task’s value or an exception thrown.

After the task’s run finishes (and regardless of an exception thrown or not), run always calls BlockManager.releaseAllLocksForTask (with the current task’s taskId).

run then always queries TaskMemoryManager for memory leaks. If there is any (i.e. the memory freed after the call is greater than 0) and spark.unsafe.exceptionOnMemoryLeak is enabled (it is not by default) with no exception having been thrown while the task was running, a SparkException is thrown:

Otherwise, if spark.unsafe.exceptionOnMemoryLeak is disabled or an exception was thrown by the task, the following ERROR message is displayed in the logs instead:

If there are any releasedLocks (after calling BlockManager.releaseAllLocksForTask earlier) and spark.storage.exceptionOnPinLeak is enabled (it is not by default) with no exception having been thrown while the task was running, a SparkException is thrown:

Otherwise, if spark.storage.exceptionOnPinLeak is disabled or an exception was thrown by the task, the following WARN message is displayed in the logs instead:

The taskFinish time which corresponds to the current time is recorded.

If the task was killed a TaskKilledException is thrown (and the TaskRunner exits).

When a task finishes successfully, it returns a value. The value is serialized (using a new instance of Serializer from SparkEnv, i.e. serializer).

The time to serialize the task’s value is tracked (using beforeSerialization and afterSerialization).

The task’s metrics are set, i.e. executorDeserializeTime, executorRunTime, jvmGCTime, and resultSerializationTime.

run collects the latest values of accumulators (as accumUpdates).

A DirectTaskResult object with the serialized result and the latest values of accumulators is created (as directResult). The object is then serialized (using the global closure Serializer).

The limit of the buffer for the serialized DirectTaskResult object is calculated (as resultSize).

The serializedResult is calculated (that soon will be sent to ExecutorBackend). It depends on the size of resultSize.

If maxResultSize is set and the size of the serialized DirectTaskResult exceeds it, the following WARN message is displayed in the logs:

The final serializedResult becomes a serialized IndirectTaskResult with a TaskResultBlockId for the task’s taskId and resultSize.

Otherwise, when maxResultSize is not positive or resultSize is smaller than maxResultSize, but greater than maxDirectResultSize, a TaskResultBlockId object for the task’s taskId is created (as blockId) and serializedDirectResult is stored as a blockId block to BlockManager with MEMORY_AND_DISK_SER storage level.

The following INFO message is printed out to the logs:

The final serializedResult becomes a serialized IndirectTaskResult with a TaskResultBlockId for the task’s taskId and resultSize.

When the two cases above do not hold, the following INFO message is printed out to the logs:

The final serializedResult becomes the serializedDirectResult (that is the serialized DirectTaskResult).

The serializedResult serialized result for the task is sent to the driver using ExecutorBackend as TaskState.FINISHED.

When the TaskRunner finishes, taskId is removed from the internal runningTasks map of the owning Executor (that ultimately cleans up any references to the TaskRunner).

kill marks the current instance of TaskRunner as killed and passes the call to kill a task on to the task itself (if available).

When executed, you should see the following INFO message in the logs:

Internally, kill enables the internal flag killed and executes its Task.kill method if a task is available.