diff options
author | Bryan Cutler <cutlerb@gmail.com> | 2016-01-08 11:08:45 -0800 |
---|---|---|
committer | Shixiong Zhu <shixiong@databricks.com> | 2016-01-08 11:08:45 -0800 |
commit | ea104b8f1ce8aa109d1b16b696a61a47df6283b2 (patch) | |
tree | 8112a6cee3eb7a0e5766426cef22dc815eadf38e | |
parent | cfe1ba56e4ab281a9e8eaf419fb7429f93c7a0ce (diff) | |
download | spark-ea104b8f1ce8aa109d1b16b696a61a47df6283b2.tar.gz spark-ea104b8f1ce8aa109d1b16b696a61a47df6283b2.tar.bz2 spark-ea104b8f1ce8aa109d1b16b696a61a47df6283b2.zip |
[SPARK-12701][CORE] FileAppender should use join to ensure writing thread completion
Changed Logging FileAppender to use join in `awaitTermination` to ensure that thread is properly finished before returning.
Author: Bryan Cutler <cutlerb@gmail.com>
Closes #10654 from BryanCutler/fileAppender-join-thread-SPARK-12701.
-rw-r--r-- | core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala | 11 |
1 files changed, 1 insertions, 10 deletions
diff --git a/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala b/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala index 14b6ba4af4..58c8560a3d 100644 --- a/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala +++ b/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala @@ -29,7 +29,6 @@ private[spark] class FileAppender(inputStream: InputStream, file: File, bufferSi extends Logging { @volatile private var outputStream: FileOutputStream = null @volatile private var markedForStop = false // has the appender been asked to stopped - @volatile private var stopped = false // has the appender stopped // Thread that reads the input stream and writes to file private val writingThread = new Thread("File appending thread for " + file) { @@ -47,11 +46,7 @@ private[spark] class FileAppender(inputStream: InputStream, file: File, bufferSi * or because of any error in appending */ def awaitTermination() { - synchronized { - if (!stopped) { - wait() - } - } + writingThread.join() } /** Stop the appender */ @@ -77,10 +72,6 @@ private[spark] class FileAppender(inputStream: InputStream, file: File, bufferSi logError(s"Error writing stream to file $file", e) } finally { closeFile() - synchronized { - stopped = true - notifyAll() - } } } |