This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7804-ff5bd7f8604366feda569ab29917bdb30a460266 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 226124bb66491b2abccb02269cf23ebaf151bb84 Author: Xinyuan Lin <[email protected]> AuthorDate: Thu Aug 20 19:46:24 2026 +0000 fix(amber, operator): keep error information on three failure paths (#7804) ### What changes were proposed in this PR? Three error paths that silently lost the information they exist to surface. All were characterized during earlier coverage work and deliberately left unpinned so a fix would not have to fight a test asserting the broken behaviour. **1. `OPVersion.getVersion` no longer returns `null` when `git log` fails.** The `GitAPIException` catch now stores the same `"N/A"` sentinel its `NullPointerException` sibling already did, so the result is non-null *and* memoized — previously the trailing `opMap.get(operatorName)` handed back `null`, which propagated into operator metadata (`LogicalOp.scala:462-466` → `OperatorMetadataGenerator.scala:171`), and every later call re-ran the failing `git log`. `printStackTrace()` is kept. That `"N/A"` is the intended sentinel is corroborated downstream: `agent-service/src/agent/util/workflow-utils.ts:193` already defends with `?? "N/A"`. **2. `InputManager.startInputPortReaderThreads` now passes the cause.** It wrapped a start failure as `new RuntimeException(s"…: ${e.getMessage}")` without `e`, discarding the original stack trace. The reader thread itself already does this correctly (`InputPortMaterializationReaderThread.scala:147`). **3. `PekkoActorRefMappingService.retrieveActorRef`'s catch no longer re-reads the failing value.** It bound `e` but never used it, instead reading `actorService.parent` again to name the parent in its warning — so a **persistently** unreachable parent threw straight out of the handler meant to contain it. The message now omits the parent and passes `e` as the log throwable, matching house style (`WorkflowActor.scala:144`, `SyncExecutionResource.scala:705`). **Why not hoist the read instead:** capturing the parent ref before the `try` was considered and rejected — the read *is* what throws, so hoisting it would let the failure escape `retrieveActorRef` entirely rather than fixing anything. ### All three are pinned, verified in both directions The production files were reverted and restored to confirm each test actually fails without its fix: | Suite | production reverted | with fixes | |---|---|---| | `OPVersionSpec` | **9 passed, 2 failed** | **11 passed** | | `InputManagerSpec` + `PekkoActorRefMappingServiceSpec` | **20 passed, 2 failed** | **22 passed** | The before-state failures are the right ones: - `null was not equal to "N/A"` (both new OPVersion cases) - `null equaled null the original failure must be attached as the cause` - `java.lang.IllegalStateException: parent is unreachable at … PekkoActorRefMappingService.retrieveActorRef` — i.e. thrown *out of* the handler, exactly the predicted failure mode ### On the tests - **OPVersion**: two cases using the existing unborn-HEAD throwaway-repo technique — the `"N/A"` fallback, and that it is *memoized* so a failing `git log` is not retried. - **InputManager**: a `NoOpReaderThread` subclass (no-op `run()`, so no storage is touched) is started and joined, then installed into the private map by reflection — the same technique as the existing `OutputPortStorageWriterThreadSpec.installWriterThread`. `Thread.start()` on a TERMINATED thread then throws `IllegalThreadStateException` deterministically, and the test asserts both the wrapping message and that `getCause` is that exception. - **PekkoActorRefMappingService**: the new case sets the failure counter to `Int.MaxValue` so *every* read throws, not just the first — the existing single-failure test could not have caught this. It also asserts the id remains askable afterwards, i.e. was not wrongly marked queried. Three stale comments were updated rather than left contradicting the new assertions: `OPVersionSpec`'s "Deliberately NOT asserted" paragraph about the null, and the defect notes on the two amber specs. ### Verification - `OPVersionSpec` 11/11; `InputManagerSpec` + `PekkoActorRefMappingServiceSpec` + `InputPortMaterializationReaderThreadSpec` 32/32 (the reader-thread spec included as the nearest neighbour to the `InputManager` change). - Specs were filtered deliberately rather than running the whole `WorkflowOperator` module, because `FileScanSourceOpExecSpec` currently aborts on Windows over a leaked handle — that is being fixed separately in #7800. - `scalafmtCheck`, `Test/scalafmtCheck` and `scalafixAll --check` pass for both modules. One pre-existing scalafix *warning* in an untouched file (`OutputManagerSpec.scala:59`, unused suppression) is unrelated. ### Any related issues, documentation, discussions? Closes #7803 ### How was this PR tested? ``` sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.metadata.OPVersionSpec" "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.messaginglayer.InputManagerSpec org.apache.texera.amber.engine.architecture.common.PekkoActorRefMappingServiceSpec" ``` ``` [info] Tests: succeeded 11, failed 0, canceled 0, ignored 0, pending 0 [info] Tests: succeeded 22, failed 0, canceled 0, ignored 0, pending 0 ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../common/PekkoActorRefMappingService.scala | 7 +- .../architecture/messaginglayer/InputManager.scala | 3 +- .../common/PekkoActorRefMappingServiceSpec.scala | 29 +++++++- .../messaginglayer/InputManagerSpec.scala | 77 +++++++++++++++++++++- .../texera/amber/operator/metadata/OPVersion.java | 4 ++ .../amber/operator/metadata/OPVersionSpec.scala | 46 +++++++++++-- 6 files changed, 153 insertions(+), 13 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala index f3fa7f3b92..223ffdf8c6 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala @@ -112,8 +112,13 @@ class PekkoActorRefMappingService(actorService: PekkoActorService) extends Amber queriedActorVirtualIdentities.add(id) } catch { case e: Throwable => + // Deliberately does not read `actorService.parent` again: that is the value whose + // failure this handler exists to contain, so re-reading it to name the parent ref + // makes a persistently unreachable parent throw straight out of the catch block. + // The exception carries the detail (including which lookup failed and why). logger.warn( - s"Failed to fetch actorRef for ${VirtualIdentityUtils.toShorterString(id)} parentRef = " + actorService.parent + s"Failed to fetch actorRef for ${VirtualIdentityUtils.toShorterString(id)} from parent", + e ) } } diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala index 4e29782943..c017eb238b 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala @@ -105,7 +105,8 @@ class InputManager( } catch { case e: Exception => throw new RuntimeException( - s"Error starting input port materialization reader thread: ${e.getMessage}" + s"Error starting input port materialization reader thread: ${e.getMessage}", + e ) } }) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala index d836008dac..db830f0b5e 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala @@ -170,9 +170,7 @@ class PekkoActorRefMappingServiceSpec val destination = ActorVirtualIdentity("unreachable-parent-destination") val waiter = TestProbe() // Pekko itself does not fail here -- `context.parent` reads a field and `!` never throws -- so - // the failure is injected, and only for the first read: the catch block's own log line reads - // `actorService.parent` a second time, which means a parent that kept failing would throw out - // of the handler that exists to contain it. + // the failure is injected, for the first read only. val actorService = new FailingParentActorService(workerId, newContext(parent)) val service = new PekkoActorRefMappingService(actorService) actorService.failuresLeft = 1 @@ -190,6 +188,31 @@ class PekkoActorRefMappingServiceSpec assert(parent.expectMsgType[GetActorRef].id == destination) } + + it should "contain a parent lookup that keeps failing, not just its first failure" in { + val parent = TestProbe() + val destination = ActorVirtualIdentity("persistently-unreachable-parent-destination") + val waiter = TestProbe() + val actorService = new FailingParentActorService(workerId, newContext(parent)) + val service = new PekkoActorRefMappingService(actorService) + // Every read of `parent` throws, which is the realistic shape of an unreachable parent: the + // condition is a property of the actor, not of one attempt. A handler that reads the failing + // value a second time (e.g. to name the parent ref in its own log line) therefore throws out + // of the very try/catch that exists to contain the failure. + actorService.failuresLeft = Int.MaxValue + + service.retrieveActorRef(destination, Set(waiter.ref)) + + parent.expectNoMessage(100.millis) + waiter.expectNoMessage(100.millis) + + // Once the parent recovers, the id must still be askable -- the contained failure may not have + // marked it queried. + actorService.failuresLeft = 0 + service.retrieveActorRef(destination, Set(waiter.ref)) + + assert(parent.expectMsgType[GetActorRef].id == destination) + } } /** Minimal actor used only to obtain a live [[ActorContext]] from Pekko TestKit. */ diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala index bcaf91e97d..2beaf4a428 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala @@ -22,12 +22,17 @@ package org.apache.texera.amber.engine.architecture.messaginglayer import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} import org.apache.texera.amber.core.workflow.PortIdentity -import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + Partitioning +} import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement +import org.apache.texera.amber.engine.architecture.worker.managers.InputPortMaterializationReaderThread import org.scalatest.flatspec.AnyFlatSpec import java.net.URI import java.util.concurrent.LinkedBlockingQueue +import scala.collection.mutable class InputManagerSpec extends AnyFlatSpec { @@ -221,4 +226,74 @@ class InputManagerSpec extends AnyFlatSpec { mgr.startInputPortReaderThreads() // must not throw succeed } + + // --------------------------------------------------------------------------- + // startInputPortReaderThreads — a start() that fails + // --------------------------------------------------------------------------- + // + // A reader thread whose body does nothing, so it can be driven to a terminated + // state without opening any materialization storage. `InputManager` only ever + // calls `start()` on it, and `Thread.start()` on a thread that is no longer NEW + // raises `IllegalThreadStateException` — a real, deterministic start failure with + // no timing window. + private class NoOpReaderThread + extends InputPortMaterializationReaderThread( + uri = new URI("file:///nowhere"), + inputMessageQueue = emptyQueue(), + workerActorId = actorId, + partitioning = BroadcastPartitioning( + batchSize = 1, + channels = Seq(channelId("upstream", "worker-1")) + ) + ) { + override def run(): Unit = () + } + + /** + * Reaches into InputManager's private inputPortMaterializationReaderThreads map, the + * way OutputPortStorageWriterThreadSpec does for OutputManager: the map is only + * populated by `addPort`, which builds real reader threads from real URIs, and the + * failure under test is in how `startInputPortReaderThreads` reports a failed start, + * not in what the thread reads. + */ + private def installReaderThreads( + manager: InputManager, + portId: PortIdentity, + threads: List[InputPortMaterializationReaderThread] + ): Unit = { + val field = classOf[InputManager] + .getDeclaredField("inputPortMaterializationReaderThreads") + field.setAccessible(true) + field + .get(manager) + .asInstanceOf[mutable.HashMap[PortIdentity, List[InputPortMaterializationReaderThread]]] + .put(portId, threads) + } + + "InputManager.startInputPortReaderThreads (failing start)" should + "wrap the failure with the original exception attached as its cause" in { + val mgr = freshManager + val alreadyStarted = new NoOpReaderThread + alreadyStarted.start() + alreadyStarted.join() // now TERMINATED, so a second start() is guaranteed to fail + installReaderThreads(mgr, PortIdentity(0), List(alreadyStarted)) + + val thrown = intercept[RuntimeException] { + mgr.startInputPortReaderThreads() + } + + // The wrapper, not the raw IllegalThreadStateException (which is itself a + // RuntimeException and would otherwise satisfy the intercept above). + assert( + thrown.getMessage.contains("Error starting input port materialization reader thread"), + s"expected the wrapping message, got: ${thrown.getMessage}" + ) + // Dropping the cause discards the only stack trace that says where the start + // failed; the message alone names neither the port nor the failing frame. + assert(thrown.getCause != null, "the original failure must be attached as the cause") + assert( + thrown.getCause.isInstanceOf[IllegalThreadStateException], + s"expected the start failure itself as the cause, got: ${thrown.getCause}" + ) + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OPVersion.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OPVersion.java index acf8828d56..bd0daf0fd8 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OPVersion.java +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OPVersion.java @@ -46,6 +46,10 @@ public class OPVersion { opMap.put(operatorName, version); } catch (GitAPIException e) { e.printStackTrace(); + // Memoize the same sentinel the NullPointerException sibling stores: the + // trailing opMap.get() would otherwise hand the caller a null version, and + // every later call for this operator would re-run the failing `git log`. + opMap.put(operatorName, "N/A"); } catch (NullPointerException e) { opMap.put(operatorName, "N/A"); } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala index a61014569a..2004bf489c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala @@ -53,9 +53,15 @@ import scala.util.Using * - a repository with history: the newest commit touching the operator's path, * memoized under the operator name; * - a repository with an unborn HEAD: `LogCommand.call()` raises `NoHeadException` - * (a `GitAPIException`) and resolution must not propagate it; + * (a `GitAPIException`), which resolution must not propagate and must answer with + * the `"N/A"` fallback; * - no handle at all: `git.log()` raises `NullPointerException` and resolution - * takes the `"N/A"` fallback. + * takes the same `"N/A"` fallback. + * + * Both failure paths are pinned to `"N/A"` rather than to null, and both are pinned as + * memoized: the version string is embedded in operator metadata, and a caller that is + * handed null has no way to tell "resolution failed" from "the field was never set", + * while an unmemoized failure makes every later call retry the same failing `git log`. * * On top of that this spec pins the memoization contract: the answer is cached per * operator name and the path is ignored on a cache hit. @@ -63,11 +69,6 @@ import scala.util.Using * Deliberately NOT covered: the static initializer itself. It runs once, before any * test can observe it, and which of its two branches executes is fixed by how the * tree was checked out — no test can flip it without mutating the JVM's environment. - * - * Deliberately NOT asserted: the value the `GitAPIException` path returns. That catch - * leaves `opMap` unpopulated, so the trailing `opMap.get(operatorName)` hands the caller - * a null version — a latent defect (the `NullPointerException` sibling stores `"N/A"`). - * Pinning the null would cement it, so only the swallow itself is asserted. */ class OPVersionSpec extends AnyFlatSpec with Matchers { @@ -200,6 +201,37 @@ class OPVersionSpec extends AnyFlatSpec with Matchers { } } + it should "fall back to \"N/A\" when the git log call fails" in { + val name = uniqueName() + withCleanCache(name) { + withTempRepo("opversion-unborn-fallback") { (handle, _) => + withGit(handle) { + // Same unborn-HEAD repository as above, so the GitAPIException catch runs. The + // answer must be the same non-null sentinel the NullPointerException sibling + // produces: this value is handed straight into operator metadata, and null + // there is indistinguishable from an unset field. + OPVersion.getVersion(name, "alpha.txt") shouldBe "N/A" + } + } + } + } + + it should "memoize the \"N/A\" fallback so a failing git log is not retried" in { + val name = uniqueName() + withCleanCache(name) { + withTempRepo("opversion-unborn-memo") { (handle, _) => + withGit(handle) { + opMap.containsKey(name) shouldBe false + OPVersion.getVersion(name, "alpha.txt") shouldBe "N/A" + // Without the memo every later call for this operator re-runs the same failing + // `git log`; the cache lookup is what makes the failure a one-off. + opMap.containsKey(name) shouldBe true + opMap.get(name) shouldBe "N/A" + } + } + } + } + // -- resolution with no usable handle ---------------------------------------- it should "fall back to \"N/A\" when the git handle is unavailable" in {
