This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 025293102d22 [SPARK-57651][CONNECT][SS][TEST] Deflake
PythonPipelineSuite and add FileStreamSink batch-commit race diagnostics
025293102d22 is described below
commit 025293102d22cbc96ce7f339ed6c36fefc9e7afb
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Wed Jun 24 19:55:03 2026 +0900
[SPARK-57651][CONNECT][SS][TEST] Deflake PythonPipelineSuite and add
FileStreamSink batch-commit race diagnostics
### What changes were proposed in this pull request?
Two changes:
1. **Deflake `PythonPipelineSuite."flow progress events have correct python
source code location"`.**
The test defined **two** flows writing to the same streaming table
`table1`: the implicit flow
created by `dp.table def table1()` and the
`dp.append_flow(target='table1') def standalone_flow1()`
append flow. Because the two flows have no data dependency on each
other, the pipeline scheduler
runs them **concurrently**, and since both write to the same file-based
streaming table they share a
single `FileStreamSink` `_spark_metadata` commit log. They then race on
the batch-0 commit in
`ManifestFileCommitProtocol.commitJob`, which throws
`IllegalStateException("Race while writing batch 0")`.
This made the test intermittently fail (e.g. it passes on most master
runs but failed in the run linked below).
The fix points `standalone_flow1` at its own dedicated streaming table
created with
`dp.create_streaming_table('st2')`, so each flow writes to a distinct
sink and there is no shared
`_spark_metadata`. The test's intent (verifying that an append flow's
Python source-code location is
propagated) is unchanged; the asserted line numbers are preserved (the
`target` is edited in place and
the `create_streaming_table` call is added below the asserted `def`
lines so nothing shifts).
Note: serializing the two flows would **not** be a correct alternative —
`FileStreamSink.addBatch`
skips a batch when `batchId <= latestBatchId`, so a serialized second
writer would silently drop its
batch rather than error. Separate sinks is the correct fix.
2. **Add diagnostics to the `ManifestFileCommitProtocol` batch-commit
race.** Previously `commitJob`
threw a bare `Race while writing batch N`. It now logs the sink
`_spark_metadata` path and the
batchId at ERROR, and the exception message names the likely cause
(multiple concurrent streaming
queries writing to the same output path). This way, if the same class of
race ever resurfaces in a
scheduled job or a real pipeline, it is diagnosable from the logs alone
without having to reproduce it.
### Why are the changes needed?
`PythonPipelineSuite` is a flaky test on CI (the race surfaces
intermittently in the
`streaming, sql-kafka-0-10, ..., connect, avro` module group). The goal is
a green, non-flaky build.
### Does this PR introduce _any_ user-facing change?
No. The only production change is a clearer error message / additional
ERROR log when the
(already-existing) batch-commit race is hit.
### How was this patch tested?
A trimmed workflow on a fork builds `sql/connect/server` and runs **only**
`PythonPipelineSuite`,
**repeated 10×** (a single green run does not prove a flaky race is gone).
All 10 runs were green
(106 tests, 0 failures each).
- ❌ Before (race observed) — apache/spark master push run, module group
incl. `connect`:
https://github.com/apache/spark/actions/runs/27961702441/job/82748952932
- ✅ After (10× green) — fork validation run, `PythonPipelineSuite` x10:
https://github.com/HyukjinKwon/spark/actions/runs/28074747108
This pull request and its description were written by Isaac.
Closes #56722 from HyukjinKwon/SPARK-57651.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/sql/connect/pipelines/PythonPipelineSuite.scala | 10 +++++++++-
.../execution/streaming/ManifestFileCommitProtocol.scala | 14 +++++++++++++-
2 files changed, 22 insertions(+), 2 deletions(-)
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index 834e2d8144e1..8eaf008c5809 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -180,6 +180,12 @@ class PythonPipelineSuite
}
test("flow progress events have correct python source code location") {
+ // `standalone_flow1` writes to its own dedicated streaming table `st2`
(rather than appending
+ // to `table1`, which already has its own implicit flow). Two flows
writing concurrently to the
+ // same file-based streaming table share a single `_spark_metadata` log
and race on the batch 0
+ // commit (`ManifestFileCommitProtocol`: "Race while writing batch 0"),
which made this test
+ // flaky. Keeping each flow on a separate destination removes the race
without changing what the
+ // test verifies (source-code-location propagation for an append flow).
val unresolvedGraph = buildGraph(pythonText = """
|@dp.table(
| comment = 'my table'
@@ -197,10 +203,12 @@ class PythonPipelineSuite
| return df.select("age")
|
|@dp.append_flow(
- | target = 'table1'
+ | target = 'st2'
|)
|def standalone_flow1():
| return spark.readStream.table('mv2')
+ |
+ |dp.create_streaming_table('st2')
|""".stripMargin)
val updateContext = TestPipelineUpdateContext(spark, unresolvedGraph,
storageRoot)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
index 66e90ec68913..804faecd37f4 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
@@ -78,7 +78,19 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
if (fileLog.add(batchId, fileStatuses)) {
logInfo(log"Committed batch ${MDC(BATCH_ID, batchId)}")
} else {
- throw new IllegalStateException(s"Race while writing batch $batchId")
+ // Reaching here means `fileLog.add` found this batchId already
committed to the sink
+ // metadata log at `path`. This is almost always two concurrent
streaming queries writing
+ // to the same output path: they share a single `_spark_metadata` log
and cannot coexist.
+ // Log the path + batchId at ERROR so a recurrence in scheduled jobs is
diagnosable from the
+ // logs alone, without re-reproducing the race.
+ // Build the message once and reuse it for both the log and the
exception so they stay in
+ // sync (see review on SPARK-57651).
+ val errorMsg = log"Race while writing batch ${MDC(BATCH_ID, batchId)} to
the file sink " +
+ log"metadata log at ${MDC(PATH, path)}: another writer already
committed this batch. " +
+ log"This usually means multiple concurrent streaming queries are
writing to the same " +
+ log"output path (they share one _spark_metadata log)."
+ logError(errorMsg)
+ throw new IllegalStateException(errorMsg.message)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
