This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new b45a8c7b4b65 [SPARK-57523][SQL] Declarative Pipelines should not retry
flows that fail due to streaming source changes
b45a8c7b4b65 is described below
commit b45a8c7b4b653d49eea8dcbaf2a07017f23fcc59
Author: YangJie <[email protected]>
AuthorDate: Thu Aug 13 14:46:37 2026 +0800
[SPARK-57523][SQL] Declarative Pipelines should not retry flows that fail
due to streaming source changes
### What changes were proposed in this pull request?
When a Declarative Pipelines streaming flow fails because its set of
sources changed since the last run, the failure is unrecoverable without a full
refresh. The retry decision in
`GraphExecution.determineFlowExecutionActionFromError` only considered the
retry count, so such a flow was retried `maxFlowRetryAttempts` times against
the same checkpoint, re-failing identically each time, and the run ended up
reporting the unrelated "has failed more than N times" reason.
This PR teaches `determineFlowExecutionActionFromError` - the method the
code already nominates as the single source of truth for retryability - to
recognize the source-change error and stop the flow immediately with a
dedicated `StreamingSourcesChanged` reason. The predicate is factored into
`PipelinesErrors.streamingSourcesChanged`, and the now-redundant special-case
branch in `checkStreamingErrorsAndRetry` is removed so the handling flows
through the single path.
The run-level failure now uses a dedicated `StreamingSourcesChangedFailure`
instead of reusing `QueryExecutionFailure`. Reusing it meant passing
`maxRetries = 0`, which prints the generic "has failed." text. That text also
collided with a real max-retries failure, since
`spark.sql.pipelines.maxFlowRetryAttempts` has no lower bound and a configured
0 takes the same branch.
### Why are the changes needed?
A streaming source-set change cannot be recovered by retrying against the
same checkpoint - every retry hits the identical `AssertionError` ("There are
[N] sources in the checkpoint offsets and now there are [M] sources requested
by the query. Cannot continue."). Retrying it wastes work, emits the same error
repeatedly, and reports a misleading terminal reason instead of the actual
cause.
### Does this PR introduce _any_ user-facing change?
Yes, in two ways. A flow whose streaming sources changed now fails
immediately instead of being retried `maxFlowRetryAttempts` times. The run also
terminates with `Run is FAILED since flow 'X' had streaming sources added or
removed. Perform a full refresh to rebuild it against the current sources.`
rather than "has failed more than N times". A full refresh is still required,
as before. This is a behavior change within the unreleased/maintained branches,
not relative to a different rel [...]
### How was this patch tested?
Extended `TriggeredGraphExecutionSuite` ("stream failure on deletes and
updates gives clear error") to assert the flow fails exactly once (not
`maxFlowRetryAttempts + 1` times) and that the run terminates with the
source-change message. Added `GraphExecutionSuite` with three session-less
tests. One checks `streamingSourcesChanged` against a wrapped source-change
error, a bare one, and errors it must not match. The other two check
`determineFlowExecutionActionFromError` on the source-c [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56588 from LuciferYang/sdp-streaming-source-change-no-retry.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit a1afa4ac6f4c8cdb153c71f1a2a88a9748dd9d53)
Signed-off-by: yangjie01 <[email protected]>
---
.../spark/sql/pipelines/graph/GraphExecution.scala | 38 +++++++--
.../sql/pipelines/graph/PipelinesErrors.scala | 50 ++++++------
.../sql/pipelines/graph/RunTerminationReason.scala | 15 ++++
.../sql/pipelines/graph/GraphExecutionSuite.scala | 90 ++++++++++++++++++++++
.../graph/TriggeredGraphExecutionSuite.scala | 13 ++++
5 files changed, 176 insertions(+), 30 deletions(-)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala
index c687c7f01ed7..db1e6c546da1 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala
@@ -261,12 +261,32 @@ object GraphExecution extends Logging {
}
/**
- * Analyze the exception thrown by flow execution and figure out if we
should retry the execution,
- * or we need to reanalyze the flow entirely to resolve issues like schema
changes.
+ * Represents that the `FlowExecution` should be stopped because a streaming
flow's set of
+ * sources changed since the last run. This is unrecoverable without a full
refresh, so the flow
+ * must not be retried regardless of the remaining retry budget.
+ */
+ private case class StreamingSourcesChanged(
+ cause: Throwable,
+ flowDisplayName: String
+ ) extends FlowExecutionStopReason {
+ override lazy val runTerminationReason: RunTerminationReason = {
+ StreamingSourcesChangedFailure(flowDisplayName, Option(cause))
+ }
+ override lazy val failureMessage: String = {
+ s"Flow '$flowDisplayName' had streaming sources added or removed. It
will not be " +
+ s"retried. Please perform a full refresh to rebuild it against the
current sources."
+ }
+ }
+
+ /**
+ * Analyze the exception thrown by flow execution and decide whether to
retry the execution or
+ * stop it. The result is either RetryFlowExecution or StopFlowExecution;
this function does not
+ * reanalyze the flow itself.
* This should be the narrow waist for all exception analysis in flow
execution.
- * TODO: currently it only handles schema change and max retries, we should
aim to extend this to
- * include other non-retryable exception as well so we can have a single
SoT for all these error
- * matching logic.
+ * Currently it handles max retries and streaming source changes; other
non-retryable errors are
+ * still routed through the retry path.
+ * TODO: extend this to include other non-retryable exceptions as well so we
can have a single
+ * SoT for all these error matching logic.
* @param ex Exception to analyze.
* @param flowDisplayName The user facing flow name with the error.
* @param currentNumTries Number of times the flow has been tried.
@@ -278,8 +298,12 @@ object GraphExecution extends Logging {
currentNumTries: => Int,
maxAllowedRetries: => Int
): FlowExecutionAction = {
- val flowExecutionNonRetryableReasonOpt = if (currentNumTries >
maxAllowedRetries) {
- Some(MaxRetryExceeded(ex, flowDisplayName, maxAllowedRetries))
+ val error = ex
+ val flowExecutionNonRetryableReasonOpt = if
(PipelinesErrors.streamingSourcesChanged(error)) {
+ // Source-set changes need a full refresh, so they are never retried.
+ Some(StreamingSourcesChanged(error, flowDisplayName))
+ } else if (currentNumTries > maxAllowedRetries) {
+ Some(MaxRetryExceeded(error, flowDisplayName, maxAllowedRetries))
} else {
None
}
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelinesErrors.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelinesErrors.scala
index b194e9c235fb..ef2ac9732c99 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelinesErrors.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelinesErrors.scala
@@ -63,6 +63,30 @@ object PipelinesErrors extends Logging {
getExceptionChain(throwable).exists(check)
}
+ /**
+ * Returns true if `ex` (or any of its causes) indicates that a streaming
flow's set of sources
+ * changed since the last run. This is unrecoverable without a full refresh,
so a flow that fails
+ * with this error must not be retried.
+ *
+ * Structured Streaming reports this as a bare `AssertionError`, so the only
signal available
+ * here is the message text, produced by the source-count assertion in
+ *
`org.apache.spark.sql.execution.streaming.checkpointing.OffsetSeq.toStreamProgress`.
Since
+ * this predicate drives the retry decision, a change to that message
silently turns these
+ * failures back into retried ones; keep the two in sync. Giving that
assertion an error
+ * condition and matching on it here would remove the coupling.
+ */
+ private[graph] def streamingSourcesChanged(ex: Throwable): Boolean = {
+ checkCauses(
+ throwable = ex,
+ check = cause => {
+ cause.isInstanceOf[AssertionError] &&
+ cause.getMessage != null &&
+ cause.getMessage.contains("sources in the checkpoint offsets and now
there are") &&
+ cause.getMessage.contains("sources requested by the query. Cannot
continue.")
+ }
+ )
+ }
+
/**
* Checks an error for streaming specific handling. This is a pretty messy
signature as a result
* of unifying some divergences between the triggered caller in
TriggeredGraphExecution and the
@@ -87,28 +111,7 @@ object PipelinesErrors extends Logging {
maxRetries: Int,
onRetry: => Unit
): Unit = {
- if (PipelinesErrors.checkCauses(
- throwable = ex,
- check = ex => {
- ex.isInstanceOf[AssertionError] &&
- ex.getMessage != null &&
- ex.getMessage.contains("sources in the checkpoint offsets and now
there are") &&
- ex.getMessage.contains("sources requested by the query. Cannot
continue.")
- }
- )) {
- val message = s"""
- |Flow '${flow.displayName}' had streaming sources added or removed.
Please perform a
- |full refresh in order to rebuild '${flow.displayName}' against the
current set of
- |sources.
- |""".stripMargin
-
- env.flowProgressEventLogger.recordFailed(
- flow = flow,
- exception = ex,
- logAsWarn = false,
- messageOpt = Option(message)
- )
- } else if (flow.once && ex == null) {
+ if (flow.once && ex == null) {
// No need to do anything if this is a ONCE flow with no exception. That
just means it's done.
} else {
val actionFromError =
GraphExecution.determineFlowExecutionActionFromError(
@@ -120,7 +123,8 @@ object PipelinesErrors extends Logging {
actionFromError match {
// Simply retry
case GraphExecution.RetryFlowExecution => onRetry
- // Schema change exception
+ // Non-retryable stop reason (max retries exceeded, streaming sources
changed, ...).
+ // When shouldRethrow is true, this rethrows so the run stops eagerly
on these reasons.
case GraphExecution.StopFlowExecution(reason) =>
val msg = reason.failureMessage
if (reason.warnInsteadOfError) {
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/RunTerminationReason.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/RunTerminationReason.scala
index c95ce6a197ee..195042286ed9 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/RunTerminationReason.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/RunTerminationReason.scala
@@ -85,6 +85,21 @@ case class QueryExecutionFailure(
}
}
+/**
+ * Indicates that a run has failed because a streaming flow's set of sources
changed since the last
+ * run. The flow is not retried, since only a full refresh can recover it.
+ */
+case class StreamingSourcesChangedFailure(
+ flowName: String,
+ override val cause: Option[Throwable])
+ extends RunFailure {
+ override def isFatal: Boolean = false
+
+ override def message: String =
+ s"Run is $terminalState since flow '$flowName' had streaming sources added
or removed. " +
+ s"Perform a full refresh to rebuild it against the current sources."
+}
+
/** Abstract class used to identify failures related to failures stopping an
operation/timeouts. */
abstract class FailureStoppingOperation extends RunFailure {
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/GraphExecutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/GraphExecutionSuite.scala
new file mode 100644
index 000000000000..f3d7108e8c22
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/GraphExecutionSuite.scala
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.SparkFunSuite
+
+/**
+ * Unit tests for the flow-execution decision logic:
`PipelinesErrors.streamingSourcesChanged` and
+ * `GraphExecution.determineFlowExecutionActionFromError`, the single source
of truth for whether a
+ * failed flow is retried.
+ *
+ * These tests are intentionally session-less - both entry points are pure
functions of a
+ * `Throwable` and two counters. End-to-end behavior (event levels, retry
counts, and the terminal
+ * run state a source change produces) is covered by
`TriggeredGraphExecutionSuite`.
+ */
+class GraphExecutionSuite extends SparkFunSuite {
+
+ /** The message Structured Streaming asserts with when a stream's source set
changes. */
+ private def sourceSetChangeMessage: String =
+ "There are [2] sources in the checkpoint offsets and now there are [3]
sources " +
+ "requested by the query. Cannot continue."
+
+ test("streamingSourcesChanged matches only a streaming source-set change
error") {
+ // Structured Streaming raises a bare AssertionError; it is usually
wrapped in another
+ // exception by the time the pipeline sees it, so the whole cause chain
has to be checked.
+ val sourceChange = new AssertionError(sourceSetChangeMessage)
+ assert(PipelinesErrors.streamingSourcesChanged(sourceChange))
+ assert(PipelinesErrors.streamingSourcesChanged(new
RuntimeException("wrapper", sourceChange)))
+ // Unrelated errors - including an AssertionError with a different message
- must not match.
+ assert(!PipelinesErrors.streamingSourcesChanged(new
RuntimeException("boom")))
+ assert(!PipelinesErrors.streamingSourcesChanged(new AssertionError("a
different assertion")))
+ }
+
+ test("determineFlowExecutionActionFromError stops on a source change before
checking retries") {
+ val sourceChange =
+ new RuntimeException("stream failed", new
AssertionError(sourceSetChangeMessage))
+
+ // The source-change check comes before the retry budget, so the flow
stops even with retries
+ // left, and the run-level reason names the source change rather than an
exhausted budget.
+ GraphExecution.determineFlowExecutionActionFromError(
+ ex = sourceChange,
+ flowDisplayName = "flow_a",
+ currentNumTries = 1,
+ maxAllowedRetries = 3) match {
+ case GraphExecution.StopFlowExecution(reason) =>
+ assert(reason.failureMessage.contains("streaming sources added or
removed"))
+ assert(
+ reason.runTerminationReason ==
+ StreamingSourcesChangedFailure("flow_a", Some(sourceChange)))
+ case other => fail(s"expected StopFlowExecution, got $other")
+ }
+ }
+
+ test("determineFlowExecutionActionFromError retries other errors until
budget is exhausted") {
+ val transient = new RuntimeException("transient")
+ // Retries remaining -> retry.
+ assert(
+ GraphExecution.determineFlowExecutionActionFromError(
+ ex = transient, flowDisplayName = "flow_a", currentNumTries = 1,
maxAllowedRetries = 3) ==
+ GraphExecution.RetryFlowExecution)
+ // Budget exhausted -> stop as max-retries-exceeded, not as a source
change.
+ GraphExecution.determineFlowExecutionActionFromError(
+ ex = transient,
+ flowDisplayName = "flow_a",
+ currentNumTries = 4,
+ maxAllowedRetries = 3) match {
+ case GraphExecution.StopFlowExecution(reason) =>
+ assert(!reason.failureMessage.contains("streaming sources added or
removed"))
+ assert(
+ reason.runTerminationReason ==
+ QueryExecutionFailure("flow_a", maxRetries = 3, Some(transient)))
+ case other => fail(s"expected StopFlowExecution, got $other")
+ }
+ }
+}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
index 313f86753222..9525a61dedd5 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
@@ -462,6 +462,8 @@ class TriggeredGraphExecutionSuite extends ExecutionTest
with SharedSparkSession
updateContext2.pipelineExecution.runPipeline()
updateContext2.pipelineExecution.awaitCompletion()
+ // A streaming source change is unrecoverable without a full refresh, so
the flow must not be
+ // retried: we should see exactly one failure rather than
maxFlowRetryAttempts + 1 of them.
assertFlowProgressEvent(
eventBuffer = updateContext2.eventBuffer,
identifier = fullyQualifiedIdentifier("input_table"),
@@ -469,6 +471,17 @@ class TriggeredGraphExecutionSuite extends ExecutionTest
with SharedSparkSession
expectedEventLevel = EventLevel.ERROR,
msgChecker = _.contains(
s"Flow '${eventLogName("input_table")}' had streaming sources added or
removed."
+ ),
+ expectedNumOfEvents = Option(1)
+ )
+
+ // The run should fail because of the source change, not because the flow
exhausted its retries.
+ assertRunProgressEvent(
+ eventBuffer = updateContext2.eventBuffer,
+ state = RunState.FAILED,
+ expectedEventLevel = EventLevel.ERROR,
+ msgChecker = _.contains(
+ s"flow '${eventLogName("input_table")}' had streaming sources added or
removed."
)
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
