[SPARK-4813][Streaming] Fix the issue that ContextWaiter didn't handle 'spurious wakeup'

Used `Condition` to rewrite `ContextWaiter` because it provides a convenient API `awaitNanos` for timeout.
One thought: what happens if `awaitTermination` times out, an error is thrown, and then we check `isTerminated`?  In that (rare) case, I guess we might miss the error-causing exception.  I suppose the caller could do some sort of double-checking, like

```scala
waiter.awaitTermination(1000)
if (!waiter.isTerminated) {
  throw Exception(...)
} else {
   waiter.awaitTermination(1)  // re-throws the error, if one occurred
}
```

Maybe this is a super-rare edge-case which we don't care about.