[SPARK-32734][SPARK-32735][DSTREAM] Fix batch submission delay caused by actions in dstream transform

[Olwn]

What changes were proposed in this pull request?

Currently dstream.getOrCompute runs at JobGenerator, which has a single thread event loop.
This pull request moves that to JobScheduler.

Why are the changes needed?

Some of our 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.
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.

Does this PR introduce any user-facing change?

No

Reproduce

// run this at Spark shell. Access Spark UI and check batch submission
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val mapped = words.transform(rdd => { Thread.sleep(10000); rdd; } )
mapped.foreachRDD(rdd => rdd.foreach(x => println(x)))

// run this at Spark shell. Access Spark UI and look streaming output and associated jobs
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val mapped = words.transform(rdd => { val c=rdd.count(); rdd.map(x => s"$c x")} )
mapped.foreachRDD(rdd => rdd.foreach(x => println(x)))

How was this patch tested?

• For the batch submission delay issue,
  I created a test ForEachDStreamSuite to make sure batch execution doesn't block batch submission.
• For Spark UI issue,
  I ran a streaming application and saw all jobs showing at batch page. A test JobSchedulerSuite is added to make sure all jobs in a batch can be associated with the BatchTime and display at Spark UI

JIRAs

https://issues.apache.org/jira/browse/SPARK-32734
https://issues.apache.org/jira/browse/SPARK-32735

[AmplabJenkins]

Can one of the admins verify this patch?

[maropu]

Thanks for the work, @Olwn. Btw, have you checked the guide? https://spark.apache.org/contributing.html You need to set correct tags in the title like [SPARK-XXXXX][some_component_tag], so I've fixed it. And, do you target to fix both issue SPARK-32734 and SPARK-32735 in a single PR? Basically, we need to make a PR for each issue.

[Olwn]

Thanks for correcting the title @maropu. I have read the contributing guide but missed this point.
Regarding SPARK-32734 and SPARK-32735, yes this PR is a small change but fixes both issues.

[Olwn]

hi @tdas , could you help review this pull request? Seems you are the main contributor of spark streaming.

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!