From 0c4367e88127254ec0ca3c54c96241ffa01eb422 Mon Sep 17 00:00:00 2001
From: anthonychengit <anthonyschenemail@gmail.com>
Date: Thu, 13 Aug 2026 16:11:58 -0700
Subject: [PATCH] fix(amber): classify wrapped sync timeouts

---
 .../web/resource/SyncExecutionResource.scala  | 15 ++++++++++-
 .../resource/SyncExecutionResourceSpec.scala  | 25 ++++++++++++++++---
 2 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
index cd528a6cf5e..7f4683893e7 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
@@ -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,
@@ -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)
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala
index 7e255374000..396e3d1b2f0 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/SyncExecutionResourceSpec.scala
@@ -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
@@ -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"))
@@ -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(