Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -256,7 +256,7 @@ class SyncExecutionResource extends LazyLogging {
.timeout(timeoutSeconds.toLong, TimeUnit.SECONDS)
.blockingGet()
} catch {
case _: java.util.concurrent.TimeoutException =>
case e: Exception if isCausedByTimeout(e) =>
killExecution(executionService)
return SyncExecutionResult(
success = false,
Expand Down Expand Up @@ -832,6 +832,19 @@ class SyncExecutionResource extends LazyLogging {
}
}

private def isCausedByTimeout(error: Throwable): Boolean = {
@scala.annotation.tailrec
def loop(current: Throwable): Boolean =
current match {
case null => false
case _: java.util.concurrent.TimeoutException => true
case throwable if throwable.getCause eq throwable => false
case throwable => loop(throwable.getCause)
}

loop(error)
}

private def hasConsoleError(consoleState: ExecutionConsoleStore): Boolean = {
consoleState.operatorConsole.values.exists { opConsole =>
opConsole.consoleMessages.exists(_.msgType == ConsoleMessageType.ERROR)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -120,10 +120,10 @@ import scala.jdk.CollectionConverters._
* - the document-reading half of `collectConsoleLogs`. Reachable by the same fixture technique
* (a console-messages document whose single column holds ASCII-serialized `ConsoleMessage`
* protos), just not done here; its database half, `getConsoleMessageUri`, is pinned below.
* - the `Observable.amb` wait and its timeout/error handlers (lines 231-277), plus the
* `ConsoleErrorDetected` / `TargetResultsReady` termination arms (284-298). Reaching them
* needs an execution that is still non-terminal when `executeWorkflowSync` looks at it, i.e.
* a live engine.
* - the `Observable.amb` wait and construction of its timeout/error responses (lines 231-277),
* plus the `ConsoleErrorDetected` / `TargetResultsReady` termination arms (284-298). Reaching
* them needs an execution that is still non-terminal when `executeWorkflowSync` looks at it,
* i.e. a live engine. Timeout cause classification is pinned independently below.
*/
class SyncExecutionResourceSpec
extends AnyFlatSpec
Expand Down Expand Up @@ -154,6 +154,7 @@ class SyncExecutionResourceSpec

private val stateToString = PrivateMethod[String](Symbol("stateToString"))
private val isTerminalState = PrivateMethod[Boolean](Symbol("isTerminalState"))
private val isCausedByTimeout = PrivateMethod[Boolean](Symbol("isCausedByTimeout"))
private val hasConsoleError = PrivateMethod[Boolean](Symbol("hasConsoleError"))
private val symmetricTruncateCellValue =
PrivateMethod[String](Symbol("symmetricTruncateCellValue"))
Expand Down Expand Up @@ -405,6 +406,22 @@ class SyncExecutionResourceSpec
resource.healthCheck shouldBe Map("status" -> "ok")
}

"isCausedByTimeout" should "detect direct and nested timeout exceptions" in {
val direct = new java.util.concurrent.TimeoutException("direct")
val nested = new RuntimeException(
"outer",
new RuntimeException("blockingGet wrapper", new java.util.concurrent.TimeoutException("wait"))
)

resource invokePrivate isCausedByTimeout(direct) shouldBe true
resource invokePrivate isCausedByTimeout(nested) shouldBe true
}

it should "reject unrelated and empty cause chains" in {
resource invokePrivate isCausedByTimeout(new RuntimeException("other")) shouldBe false
resource invokePrivate isCausedByTimeout(null.asInstanceOf[Throwable]) shouldBe false
}

"stateToString" should "name every aggregated state and fall back to Unknown" in {
// The whole mapping in one assertion: a swapped or dropped case shows up as a diff.
val named = List(
Expand Down
Loading