LuciferYang opened a new pull request, #56588:
URL: https://github.com/apache/spark/pull/56588
### 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.
### 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. A flow that fails because its streaming sources changed now fails
immediately without retrying, and the run terminates with the source-change
failure 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 released version.
### 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 no longer terminates with
"has failed more than N times". The full suite (15 tests) passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
