Github user jose-torres commented on a diff in the pull request:
https://github.com/apache/spark/pull/21400#discussion_r191278390
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
---
@@ -225,6 +226,40 @@ class ContinuousSuite extends ContinuousSuiteBase {
val results = spark.read.table("noharness").collect()
assert(Set(0, 1, 2, 3).map(Row(_)).subsetOf(results.toSet))
}
+
+ test("SPARK-24351: check offsetLog/commitLog retained in the checkpoint
directory") {
+ withTempDir { checkpointDir =>
+ val input = ContinuousMemoryStream[Int]
+ val df = input.toDF().mapPartitions(iter => {
+ // Sleep the task thread for 300 ms to make sure epoch processing
time 3 times
+ // longer than epoch creating interval. So the gap between last
committed
+ // epoch and currentBatchId grows over time.
+ Thread.sleep(300)
--- End diff --
I think this is a reasonable use of sleep(), since we're checking to ensure
that this sleep() specifically doesn't cause anything bad to happen.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
