Fix race condition in CheckpointWriter shutdown.

JoshRosen · JoshRosen · commit 61a388a62c37 · 2016-04-26T14:08:28.000-07:00
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala b/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala
@@ -184,8 +184,7 @@ class CheckpointWriter(
   val executor = Executors.newFixedThreadPool(1)
   val compressionCodec = CompressionCodec.createCodec(conf)
   private var stopped = false
-  private var _fs: FileSystem = _
-
+  @volatile private[this] var fs: FileSystem = null
   @volatile private var latestCheckpointTime: Time = null
 
   class CheckpointWriteHandler(
@@ -196,6 +195,9 @@ class CheckpointWriter(
       if (latestCheckpointTime == null || latestCheckpointTime < checkpointTime) {
         latestCheckpointTime = checkpointTime
       }
+      if (fs == null) {
+        fs = new Path(checkpointDir).getFileSystem(hadoopConf)
+      }
       var attempts = 0
       val startTime = System.currentTimeMillis()
       val tempFile = new Path(checkpointDir, "temp")
@@ -263,7 +265,7 @@ class CheckpointWriter(
           case ioe: IOException =>
             logWarning("Error in attempt " + attempts + " of writing checkpoint to "
               + checkpointFile, ioe)
-            reset()
+            fs = null
         }
       }
       logWarning("Could not write checkpoint for time " + checkpointTime + " to file "
@@ -297,15 +299,6 @@ class CheckpointWriter(
       ", waited for " + (endTime - startTime) + " ms.")
     stopped = true
   }
-
-  private def fs = synchronized {
-    if (_fs == null) _fs = new Path(checkpointDir).getFileSystem(hadoopConf)
-    _fs
-  }
-
-  private def reset() = synchronized {
-    _fs = null
-  }
 }
 
 
