TaskRunner

TaskRunner takes the following when created:

TaskRunner initializes the internal registries and counters.

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 executed, run initializes threadId as the current thread identifier (using Java’s Thread)

run then sets the name of the current thread as threadName (using Java’s Thread).

run creates a TaskMemoryManager (using the current MemoryManager and taskId).

run starts tracking the time to deserialize a task.

run sets the current thread’s context classloader (with replClassLoader).

run creates a closure Serializer.

You should see the following INFO message in the logs:

run notifies ExecutorBackend that taskId is in TaskState.RUNNING state.

run computes startGCTime.

run updates dependencies.

run deserializes the task (using the context class loader) and sets its localProperties and TaskMemoryManager. run sets the task internal reference to hold the deserialized task.

If killed flag is enabled, run throws a TaskKilledException.

You should see the following DEBUG message in the logs:

run notifies MapOutputTracker about the epoch of the task.

run records the current time as the task’s start time (as taskStart).

run runs the task (with taskAttemptId as taskId, attemptNumber from TaskDescription, and metricsSystem as the current MetricsSystem).

After the task’s run has finished (inside the “finally” block of the “monitored” block), run requests BlockManager to release all locks of the task (for the task’s taskId). The locks are later used for lock leak detection.

run then requests TaskMemoryManager to clean up allocated memory (that helps finding memory leaks).

If run detects memory leak of the managed memory (i.e. the memory freed is greater than 0) and spark.unsafe.exceptionOnMemoryLeak Spark property is enabled (it is not by default) and no exception was reported while the task ran, run reports a SparkException:

Otherwise, if spark.unsafe.exceptionOnMemoryLeak is disabled, you should see the following ERROR message in the logs instead:

If run detects lock leaking (i.e. the number of locks released) and spark.storage.exceptionOnPinLeak Spark property is enabled (it is not by default) and no exception was reported while the task ran, run reports a SparkException:

Otherwise, if spark.storage.exceptionOnPinLeak is disabled or the task reported an exception, you should see the following INFO message in the logs instead:

Rigth after the “monitored” block, run records the current time as the task’s finish time (as taskFinish).

If the task was killed (while it was running), run reports a TaskKilledException (and the TaskRunner exits).

run creates a Serializer and serializes the task’s result. run measures the time to serialize the result.

run records the task metrics:

run collects the latest values of internal and external accumulators used in the task.

run creates a DirectTaskResult (with the serialized result and the latest values of accumulators).

run serializes the DirectTaskResult and gets the byte buffer’s limit.

run selects the proper serialized version of the result before sending it to ExecutorBackend.

run branches off based on the serialized DirectTaskResult byte buffer’s limit.

When maxResultSize is greater than 0 and the serialized DirectTaskResult buffer limit exceeds it, the following WARN message is displayed in the logs:

In this case, run creates a IndirectTaskResult (with a TaskResultBlockId for the task’s taskId and resultSize) and serializes it.

When maxResultSize is not positive or resultSize is smaller than maxResultSize but greater than maxDirectResultSize, run creates a TaskResultBlockId for the task’s taskId and stores the serialized DirectTaskResult in BlockManager (as the TaskResultBlockId with MEMORY_AND_DISK_SER storage level).

You should see the following INFO message in the logs:

In this case, run creates a IndirectTaskResult (with a TaskResultBlockId for the task’s taskId and resultSize) and serializes it.

When the two cases above do not hold, you should see the following INFO message in the logs:

run uses the serialized DirectTaskResult byte buffer as the final serializedResult.

run notifies ExecutorBackend that taskId is in TaskState.FINISHED state with the serialized result and removes taskId from the owning executor’s runningTasks registry.

When run catches a exception while executing the task, run acts according to its type (as presented in the following “run’s Exception Cases” table and the following sections linked from the table).

kill marks the TaskRunner as killed and kills the task (if available and not finished already).

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