This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 226124bb66 fix(amber, operator): keep error information on three
failure paths (#7804)
226124bb66 is described below
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 {
