diff options
| author | Ryan Williams <ryan.blake.williams@gmail.com> | 2015-02-06 12:22:25 +0000 |
|---|---|---|
| committer | Sean Owen <sowen@cloudera.com> | 2015-02-06 12:22:25 +0000 |
| commit | 37d35ab53b82916e4bf0eeb788044bfc33f329df (patch) | |
| tree | 12cf814e5404a8109025ee8140d4d2f19b239d63 /core/src/main/scala/org/apache/spark/executor/Executor.scala | |
| parent | cf6778e8d8e8d7e0b4e7b17637ad812624a937dd (diff) | |
| download | spark-37d35ab53b82916e4bf0eeb788044bfc33f329df.tar.gz spark-37d35ab53b82916e4bf0eeb788044bfc33f329df.tar.bz2 spark-37d35ab53b82916e4bf0eeb788044bfc33f329df.zip | |
[SPARK-5416] init Executor.threadPool before ExecutorSource
Some ExecutorSource metrics can NPE by attempting to reference the
threadpool otherwise.
Author: Ryan Williams <ryan.blake.williams@gmail.com>
Closes #4212 from ryan-williams/threadpool and squashes the following commits:
236f2ad [Ryan Williams] init Executor.threadPool before ExecutorSource
Diffstat (limited to 'core/src/main/scala/org/apache/spark/executor/Executor.scala')
| -rw-r--r-- | core/src/main/scala/org/apache/spark/executor/Executor.scala | 6 |
1 files changed, 3 insertions, 3 deletions
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala index 312bb3a1da..5141483d1e 100644 --- a/core/src/main/scala/org/apache/spark/executor/Executor.scala +++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala @@ -75,6 +75,9 @@ private[spark] class Executor( Thread.setDefaultUncaughtExceptionHandler(SparkUncaughtExceptionHandler) } + // Start worker thread pool + val threadPool = Utils.newDaemonCachedThreadPool("Executor task launch worker") + val executorSource = new ExecutorSource(this, executorId) if (!isLocal) { @@ -101,9 +104,6 @@ private[spark] class Executor( // Limit of bytes for total size of results (default is 1GB) private val maxResultSize = Utils.getMaxResultSize(conf) - // Start worker thread pool - val threadPool = Utils.newDaemonCachedThreadPool("Executor task launch worker") - // Maintains the list of running tasks. private val runningTasks = new ConcurrentHashMap[Long, TaskRunner] |