[SPARK-12122][STREAMING] Prevent batches from being submitted twice after recovering StreamingContext from checkpoint

[tdas]

No description provided.

[tdas]

@zsxwing Please check this.
I think this problem has been caused by the #9707

[SparkQA]

Test build #2161 has finished for PR 10127 at commit d904b25.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #47134 has finished for PR 10127 at commit d904b25.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2162 has finished for PR 10127 at commit d904b25.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class CrossValidator @Since(\"1.2.0\") (@Since(\"1.4.0\") override val uid: String)\n * class ParamGridBuilder @Since(\"1.2.0\")\n * class TrainValidationSplit @Since(\"1.5.0\") (@Since(\"1.5.0\") override val uid: String)\n

[zsxwing]

Could you clarify why pendingTimes may contain restartTime?

[tdas]

Explained offline:

The restart time is always checkpointTime+1 (assuming batch duration = 1). However, pending times can already have batches >= checkpointTime+1. This can cause timesToReschedule to have batches >= checkpointTime+1, which will be explicitly submitted, and then resubmitted through the timer.

[zsxwing]

LGTM

[SparkQA]

Test build #2164 has finished for PR 10127 at commit d904b25.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2165 has finished for PR 10127 at commit d904b25.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #47163 has finished for PR 10127 at commit fe69fbf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.