diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala index 02871e64aa761..22b64798f598f 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala @@ -418,21 +418,9 @@ class TriggeredGraphExecution( return RunCompletion() } - val executionFailureOpt = failureTracker.iterator - .map { - case (flowIdentifier, failureInfo) => - ( - graphForExecution.flow(flowIdentifier), - failureInfo.lastException, - failureInfo.lastExceptionAction - ) - } - .collectFirst { - case (_, _, GraphExecution.StopFlowExecution(reason)) => - reason.runTerminationReason - } - - executionFailureOpt.getOrElse(UnexpectedRunFailure()) + TriggeredGraphExecution + .chooseRunTerminationReason(failureTracker.iterator) + .getOrElse(UnexpectedRunFailure()) } } @@ -450,6 +438,23 @@ case class TriggeredFailureInfo( object TriggeredGraphExecution { + /** + * Picks the run-termination reason from the flows whose execution was stopped because they + * exhausted their retries. Several flows can stop a run and `failures` comes from an unordered + * map, so the earliest failure is chosen - ties broken by flow name - to keep the reported + * reason stable across otherwise-identical runs. + */ + private[graph] def chooseRunTerminationReason( + failures: Iterator[(TableIdentifier, TriggeredFailureInfo)]): Option[RunTerminationReason] = { + failures + .collect { + case (id, TriggeredFailureInfo(ts, _, _, GraphExecution.StopFlowExecution(r))) => + (ts, id.unquotedString, r.runTerminationReason) + } + .minByOption { case (ts, flowName, _) => (ts, flowName) } + .map { case (_, _, reason) => reason } + } + // All possible states of a data stream for a flow sealed trait StreamState object StreamState { 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 57baf4c2d5b11..db8c368ca89df 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 @@ -1060,4 +1060,53 @@ class TriggeredGraphExecutionSuite extends ExecutionTest with SharedSparkSession assert(warnCount == 2 && errorCount == 1) } + + /** A non-retryable (retries exhausted) failure, i.e. one that stops the run. */ + private def stopFailure(ts: Long, flowName: String, cause: Throwable): TriggeredFailureInfo = { + // currentNumTries > maxAllowedRetries yields a StopFlowExecution. + val action = GraphExecution.determineFlowExecutionActionFromError( + ex = cause, + flowDisplayName = flowName, + currentNumTries = 2, + maxAllowedRetries = 1) + TriggeredFailureInfo(ts, numFailures = 2, lastException = cause, lastExceptionAction = action) + } + + test("chooseRunTerminationReason surfaces the earliest non-retryable failure deterministically") { + val earliestCause = new RuntimeException("earliest") + // A retryable failure with an earlier timestamp must be ignored: it does not stop the run. + val retryable = { + val cause = new RuntimeException("retryable") + val action = GraphExecution.determineFlowExecutionActionFromError( + ex = cause, flowDisplayName = "flow_retry", currentNumTries = 1, maxAllowedRetries = 3) + TableIdentifier("flow_retry") -> TriggeredFailureInfo( + 50, numFailures = 1, lastException = cause, lastExceptionAction = action) + } + val entries = Seq( + TableIdentifier("flow_c") -> stopFailure(300, "flow_c", new RuntimeException("c")), + TableIdentifier("flow_b") -> stopFailure(100, "flow_b", earliestCause), + TableIdentifier("flow_a") -> stopFailure(200, "flow_a", new RuntimeException("a")), + retryable) + + // The earliest non-retryable failure (flow_b @ 100) wins regardless of iteration order. + Seq(entries, entries.reverse).foreach { ordered => + assert( + TriggeredGraphExecution.chooseRunTerminationReason(ordered.iterator) + .contains(QueryExecutionFailure("flow_b", 1, Some(earliestCause)))) + } + } + + test("chooseRunTerminationReason breaks ties between equal timestamps by flow name") { + val aCause = new RuntimeException("a") + val entries = Seq( + TableIdentifier("flow_z") -> stopFailure(100, "flow_z", new RuntimeException("z")), + TableIdentifier("flow_a") -> stopFailure(100, "flow_a", aCause)) + assert( + TriggeredGraphExecution.chooseRunTerminationReason(entries.iterator) + .contains(QueryExecutionFailure("flow_a", 1, Some(aCause)))) + } + + test("chooseRunTerminationReason has no reason when no flow stopped the run") { + assert(TriggeredGraphExecution.chooseRunTerminationReason(Iterator.empty).isEmpty) + } }