[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.
Another option would be to just add a new `def isTerminated: Boolean` method, which would let users write code like

```
waiter.awaitTermination(1000)
if (!waiter.isTerminated) {
  throw Exception(...)
}
```