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 <[email protected]>
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 <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 850a41e0623834c9b7eb5457692f08b61b8d770a)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
