@zsxwing we have to keep `waitUntilEmpty` in LiveListenerBus.scala (unfortunately) because DAGSchedulerSuite also uses it. I started a similar solution at #544. Maybe we can move the discussion there.