You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/06/24 10:40:12 UTC
(spark) branch branch-4.x updated: [SPARK-57652][SS][TEST] Deflake snapshotStartBatchId-with-transformWithState StateDataSource test
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 8b62d3f21e97 [SPARK-57652][SS][TEST] Deflake snapshotStartBatchId-with-transformWithState StateDataSource test
8b62d3f21e97 is described below
commit 8b62d3f21e97b7d28a1163c2f41b3ab497191a65
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Wed Jun 24 19:39:42 2026 +0900
[SPARK-57652][SS][TEST] Deflake snapshotStartBatchId-with-transformWithState StateDataSource test
### What changes were proposed in this pull request?
Replace the fixed `Thread.sleep(5000)` in the `"snapshotStartBatchId with transformWithState"` test of `StateDataSourceTransformWithStateSuite` with a deterministic `eventually(...)` wait that polls until the RocksDB snapshot files the `snapshotStartBatchId` reader needs (snapshot version 2 for the partitions read) have actually been uploaded by the asynchronous maintenance thread.
The root cause is known, but to guard against regressions the timeout assertion now prints the actual state-directory contents, so a recurrence in scheduled jobs is immediately diagnosable (snapshot still pending vs. cleaned up vs. wrong path) rather than a bare failure.
### Why are the changes needed?
The snapshot upload is asynchronous (background maintenance thread), so the fixed sleep is racy under CI load. When it is too short, the snapshot `.zip` is not yet uploaded and the reader fails on the scheduled **Maven (Scala 2.13, JDK 21)** and **JDK 25** builds:
```
[CANNOT_LOAD_STATE_STORE.UNCATEGORIZED] An error occurred during loading state.
Caused by: java.io.FileNotFoundException: .../state/0/1/2.zip does not exist
```
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
- **Before (failing job):** [`Build / Maven (Scala 2.13, JDK 21)` → `sql#core - slow tests`](https://github.com/apache/spark/actions/runs/28048347820/job/83041196522) — `snapshotStartBatchId with transformWithState ... *** FAILED ***` (`FileNotFoundException ... 2.zip`).
- **After (passing, run 10x to confirm the flake is gone):** [✅ 10/10 passed](https://github.com/HyukjinKwon/spark/actions/runs/28074724227/job/83116481122) — the test was executed 10 consecutive times in one sbt session, all green.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56721 from HyukjinKwon/SPARK-57652-tws-deflake.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
(cherry picked from commit 850a41e0623834c9b7eb5457692f08b61b8d770a)
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
.../StateDataSourceTransformWithStateSuite.scala | 32 ++++++++++++++++++++--
1 file changed, 30 insertions(+), 2 deletions(-)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
index d7f28b79acff..f861388605e0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala
@@ -20,6 +20,7 @@ import java.io.File
import java.time.Duration
import org.apache.hadoop.conf.Configuration
+import org.scalatest.time.SpanSugar._
import org.apache.spark.io.CompressionCodec
import org.apache.spark.sql.{Encoders, Row}
@@ -1050,20 +1051,47 @@ class StateDataSourceTransformWithStateSuite extends StateStoreMetricsTest
.transformWithState(new AggregationStatefulProcessor(),
TimeMode.None(),
OutputMode.Append())
+
+ // Block until the async maintenance thread has written the RocksDB snapshot `.zip` for the
+ // given state-store `version` (matching both `<v>.zip` and checkpoint-v2 `<v>_<uid>.zip`)
+ // on each of `partitions`. Used right after `version` is committed (while it is the current
+ // version), since maintenance only ever snapshots the current version. On timeout it prints
+ // the actual directory contents so a recurrence in scheduled jobs is diagnosable.
+ def waitForStateSnapshot(version: Long, partitions: Seq[Int]): StreamAction = Execute { _ =>
+ val opStateDir = new File(tmpDir, "state/0")
+ eventually(timeout(60.seconds), interval(100.milliseconds)) {
+ partitions.foreach { partition =>
+ val partitionDir = new File(opStateDir, partition.toString)
+ val files = Option(partitionDir.listFiles())
+ .map(_.map(_.getName).sorted.toSeq).getOrElse(Seq.empty)
+ val snapshotUploaded = files.exists(_.matches(s"$version(_.*)?\\.zip"))
+ assert(snapshotUploaded,
+ s"Snapshot (version $version) for partition $partition was not uploaded in time. " +
+ s"Contents of $partitionDir: ${files.mkString("[", ", ", "]")}")
+ }
+ }
+ }
+
testStream(query)(
StartStream(checkpointLocation = tmpDir.getCanonicalPath),
AddData(inputData, (1, 1L), (2, 2L), (3, 3L), (4, 4L)),
ProcessAllAvailable(),
AddData(inputData, (5, 1L), (6, 2L), (7, 3L), (8, 4L)),
ProcessAllAvailable(),
+ // State version 2 is now the current version. The snapshotStartBatchId=1 reader below
+ // needs the version-2 snapshot, and the asynchronous maintenance thread only creates a
+ // snapshot for the *current* version. So block here, while version 2 is still current,
+ // until that snapshot has actually been written - this deterministically forces it to
+ // exist. (A fixed sleep at the *end* is flaky: once later batches advance the current
+ // version, maintenance never goes back to snapshot version 2, so it may never appear and
+ // the reader fails with FileNotFoundException on `2.zip`.)
+ waitForStateSnapshot(version = 2, partitions = Seq(1, 4)),
AddData(inputData, (9, 1L), (10, 2L), (11, 3L), (12, 4L)),
ProcessAllAvailable(),
AddData(inputData, (13, 1L), (14, 2L), (15, 3L), (16, 4L)),
ProcessAllAvailable(),
AddData(inputData, (17, 1L), (18, 2L), (19, 3L), (20, 4L)),
ProcessAllAvailable(),
- // Ensure that we get a chance to upload created snapshots
- Execute { _ => Thread.sleep(5000) },
StopStream
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org