[SPARK-14930][SPARK-13693] Fix race condition in CheckpointWriter.stop()

[JoshRosen]

CheckpointWriter.stop() is prone to a race condition: if one thread calls stop() right as a checkpoint write task begins to execute, that write task may become blocked when trying to access fs, the shared Hadoop FileSystem, since both the fs getter and stop method synchronize on the same lock. Here's a thread-dump excerpt which illustrates the problem:

"pool-31-thread-1" #156prio=5os_prio=31tid=0x00007fea02cd2000nid=0x5c0bwaitingformonitorentry [0x000000013bc4c000]
java.lang.Thread.State: BLOCKED (onobjectmonitor)
atorg.apache.spark.streaming.CheckpointWriter.org$apache$spark$streaming$CheckpointWriter$$fs(Checkpoint.scala:302)
- waitingtolock <0x00000007bf53ee78> (aorg.apache.spark.streaming.CheckpointWriter)
atorg.apache.spark.streaming.CheckpointWriter$CheckpointWriteHandler.run(Checkpoint.scala:224)
atjava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
atjava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
atjava.lang.Thread.run(Thread.java:745)
"pool-1-thread-1-ScalaTest-running-MapWithStateSuite" #11prio=5os_prio=31tid=0x00007fe9ff879800nid=0x5703waitingoncondition [0x000000012e54c000]
java.lang.Thread.State: TIMED_WAITING (parking)
atsun.misc.Unsafe.park(NativeMethod)
- parkingtowaitfor <0x00000007bf564568> (ajava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
atjava.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
atjava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
atjava.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1465)
atorg.apache.spark.streaming.CheckpointWriter.stop(Checkpoint.scala:291)
- locked <0x00000007bf53ee78> (aorg.apache.spark.streaming.CheckpointWriter)
atorg.apache.spark.streaming.scheduler.JobGenerator.stop(JobGenerator.scala:159)
- locked <0x00000007bf53ea90> (aorg.apache.spark.streaming.scheduler.JobGenerator)
atorg.apache.spark.streaming.scheduler.JobScheduler.stop(JobScheduler.scala:115)
- locked <0x00000007bf53d3f0> (aorg.apache.spark.streaming.scheduler.JobScheduler)
atorg.apache.spark.streaming.StreamingContext$$anonfun$stop$1.apply$mcV$sp(StreamingContext.scala:680)
atorg.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1219)
atorg.apache.spark.streaming.StreamingContext.stop(StreamingContext.scala:679)
- locked <0x00000007bf516a70> (aorg.apache.spark.streaming.StreamingContext)
atorg.apache.spark.streaming.StreamingContext.stop(StreamingContext.scala:644)
- locked <0x00000007bf516a70> (aorg.apache.spark.streaming.StreamingContext)
[...]

We can fix this problem by having stop and fs be synchronized on different locks: the synchronization on stop only needs to guard against multiple threads calling stop at the same time, whereas the synchronization on fs is only necessary for cross-thread visibility. There's only ever a single active checkpoint writer thread at a time, so we don't need to guard against concurrent access to fs. Thus, fs can simply become a @volatile var, similar to lastCheckpointTime.

This change should fix SPARK-13693, a flaky MapWithStateSuite test suite which has recently been failing several times per day. It also results in a huge test speedup: prior to this patch, MapWithStateSuite took about 80 seconds to run, whereas it now runs in less than 10 seconds. For the streaming project's tests as a whole, they now run in ~220 seconds vs. ~354 before.

/cc @zsxwing and @tdas for review.

[SparkQA]

Test build #57030 has finished for PR 12712 at commit 61a388a.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Jenkins, retest this please.

[SparkQA]

Test build #57040 has finished for PR 12712 at commit 61a388a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[marmbrus]

LGTM, thanks for fixing this!

[srowen]

Does this even need to be a member? it looks like it's used entirely in one method.

[JoshRosen]

It's actually referenced by inner classes. I think the idea here is that we'll avoid calling .getFileSystem too much by caching the result, but will have the ability to clear the cache in case the cached FileSystem seems to be in a bad state.

[zsxwing]

minor: fs and latestCheckpointTime don't need to be volatile as we use Executors.newFixedThreadPool(1) here. We should also give the thread a name for better debugging.

[tdas]

LGTM. Merging to master