RDD actions in DStream.transfrom delays batch submission

Issue

Some spark applications have batch creation delay after running for some time. For instance, Batch 10:03 is submitted at 10:06. In spark UI, the latest batch doesn't match current time.
 

Investigation

We observe such applications share a commonality that rdd actions exist in dstream.transfrom. Those actions will be executed in dstream.compute, which is called by JobGenerator. JobGenerator runs with a single thread event loop so any synchronized operations will block event processing.

Proposal

delegate dstream.compute to JobSchduler

// class ForEachDStream

override def generateJob(time: Time): Option[Job] = {
  val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
    parent.getOrCompute(time).foreach(rdd => foreachFunc(rdd, time))
  }
  Some(new Job(time, jobFunc))
}