This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 87ac6bd8bf61 [SPARK-57652][4.2][SS][TEST] Deflake
snapshotStartBatchId-with-transformWithState StateDataSource test
87ac6bd8bf61 is described below
commit 87ac6bd8bf61fd5a1855caac5a6952fd7bffffdd
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Fri Jun 26 16:09:36 2026 +0900
[SPARK-57652][4.2][SS][TEST] Deflake
snapshotStartBatchId-with-transformWithState StateDataSource test
### What changes were proposed in this pull request?
Backport [SPARK-57652] (master commit
`850a41e0623834c9b7eb5457692f08b61b8d770a`)
to `branch-4.2`. It replaces the fixed `Thread.sleep(5000)` in the
`"snapshotStartBatchId with transformWithState"` test of
`StateDataSourceTransformWithStateSuite` with a deterministic
`waitForStateSnapshot`
helper that blocks (via `eventually`) until the RocksDB snapshot `.zip` for
the
needed state-store version has actually been uploaded — called immediately
after
that version is committed, while it is still the current version.
### Why are the changes needed?
`branch-4.2` does not have the deflake fix, so the test is still flaky
there. It
fails on the scheduled **Maven (branch-4.2, Scala 2.13, JDK 21)** `sql#core
-
extended tests` job (run 28189497187), in
`StateDataSourceTransformWithStateSuiteWithRowChecksum`:
```
- snapshotStartBatchId with transformWithState (with changelog
checkpointing) (encoding = unsaferow) *** FAILED ***
org.apache.spark.SparkException: [CANNOT_LOAD_STATE_STORE.UNCATEGORIZED]
An error occurred during loading state.
Caused by: java.io.FileNotFoundException: .../state/0/1/2.zip does not
exist
```
The maintenance thread only ever snapshots the **current** state-store
version, so
the version-2 snapshot the `snapshotStartBatchId=1` reader needs can only be
created while version 2 is current. A `Thread.sleep` at the end of the
stream is
racy: once later batches advance the current version, maintenance never
goes back
to snapshot version 2, so `2.zip` may never appear. The backported helper
waits
for the snapshot while version 2 is still current, deterministically
forcing it to
exist.
### Does this PR introduce _any_ user-facing change?
No, test only.
### How was this patch tested?
Backported test, run on the branch-4.2 Maven JDK21 `sql#core - extended
tests`
job (see CI evidence). Clean cherry-pick of the master commit.
### CI evidence
- Before (red): https://github.com/apache/spark/actions/runs/28189497187
(branch-4.2 Maven JDK21, `sql#core - extended tests` —
`CANNOT_LOAD_STATE_STORE` / `2.zip does not exist`)
- After (green):
https://github.com/HyukjinKwon/spark/actions/runs/28205027037 (fork, branch-4.2
Maven JDK21, matrix narrowed to `sql#core - extended tests`)
### Was this patch authored or co-authored using generative AI tooling?
Yes, drafted with assistance from Isaac.
Closes #56791 from HyukjinKwon/ci-fix/agent5-tws-snapshot-b42.
Authored-by: Hyukjin Kwon <[email protected]>
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 3b0279b5e6f3..41f66807bef3 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]
