This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch release/v1.2
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 203286c1039e4d2e8c88c14883f28da5730ab3d3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Aug 23 04:35:37 2026 +0000
fix(amber, operator, v1.2): keep error information on three failure paths
(#7809)
### What changes were proposed in this PR?
Automated backport of #7804 to `release/v1.2`.
Source: 226124bb66491b2abccb02269cf23ebaf151bb84 · [automation
run](https://github.com/apache/texera/actions/runs/32411776055)
### Any related issues, documentation, discussions?
Backport of #7804. Originally linked #7803.
### How was this PR tested?
Release-branch CI runs on this branch once the conflicts are resolved
and this PR is marked ready for review.
### Was this PR authored or co-authored using generative AI tooling?
No.
---------
Co-authored-by: Xinyuan Lin <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../common/PekkoActorRefMappingService.scala | 7 +-
.../architecture/messaginglayer/InputManager.scala | 3 +-
.../common/PekkoActorRefMappingServiceSpec.scala | 236 ++++++++++++++++
.../messaginglayer/InputManagerSpec.scala | 299 ++++++++++++++++++++
.../texera/amber/operator/metadata/OPVersion.java | 4 +
.../amber/operator/metadata/OPVersionSpec.scala | 309 +++++++++++++++++++++
6 files changed, 856 insertions(+), 2 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 323435891e..e8d5daee92 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
new file mode 100644
index 0000000000..e607e1071c
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.architecture.common
+
+import org.apache.pekko.actor.{Actor, ActorContext, ActorRef, ActorSystem,
Props}
+import org.apache.pekko.testkit.{TestActorRef, TestKit, TestProbe}
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity,
ChannelIdentity}
+import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{
+ CreditRequest,
+ GetActorRef,
+ NetworkMessage,
+ RegisterActorRef
+}
+import org.apache.texera.amber.engine.common.ambermessage.{DataFrame,
WorkflowFIFOMessage}
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration.DurationInt
+
+/**
+ * Unit tests for [[PekkoActorRefMappingService]].
+ *
+ * These tests use a minimal actor spawned by Pekko TestKit to supply the
+ * live [[ActorContext]] that [[PekkoActorService]] eagerly dereferences.
+ * Spawning it as a child of a [[TestProbe]] also makes the service's parent
+ * lookup observable.
+ */
+class PekkoActorRefMappingServiceSpec
+ extends TestKit(ActorSystem("PekkoActorRefMappingServiceSpec"))
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll {
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ private val upstreamId = ActorVirtualIdentity("upstream")
+ private val workerId = ActorVirtualIdentity("mapping-service-worker")
+ private val contextCounter = new java.util.concurrent.atomic.AtomicInteger(0)
+
+ private def channelTo(destination: ActorVirtualIdentity): ChannelIdentity =
+ ChannelIdentity(
+ fromWorkerId = upstreamId,
+ toWorkerId = destination,
+ isControl = true
+ )
+
+ private def networkMessageTo(
+ destination: ActorVirtualIdentity,
+ messageId: Long,
+ sequenceNumber: Long
+ ): NetworkMessage =
+ NetworkMessage(
+ messageId,
+ WorkflowFIFOMessage(channelTo(destination), sequenceNumber,
DataFrame(Array.empty))
+ )
+
+ private def newContext(parent: TestProbe): ActorContext =
+ TestActorRef[ActorRefMappingServiceContextHolder](
+ Props(new ActorRefMappingServiceContextHolder),
+ parent.ref,
+ s"actor-ref-mapping-context-${contextCounter.incrementAndGet()}"
+ ).underlyingActor.context
+
+ private def newActorService(
+ id: ActorVirtualIdentity,
+ parent: TestProbe
+ ): PekkoActorService = new PekkoActorService(id, newContext(parent))
+
+ "askForCredit" should "forward a request only when the destination ref is
known" in {
+ val parent = TestProbe()
+ val destination = TestProbe()
+ val unknownDestination = ActorVirtualIdentity("unknown-credit-destination")
+ val knownDestination = ActorVirtualIdentity("known-credit-destination")
+ val service = new PekkoActorRefMappingService(newActorService(workerId,
parent))
+
+ service.askForCredit(channelTo(unknownDestination))
+ parent.expectNoMessage(100.millis)
+
+ service.registerActorRef(knownDestination, destination.ref)
+ val channel = channelTo(knownDestination)
+ service.askForCredit(channel)
+
+ destination.expectMsg(CreditRequest(channel))
+ }
+
+ "forwardToActor" should "stash unknown messages and ask the parent for their
ref once" in {
+ val parent = TestProbe()
+ val destination = ActorVirtualIdentity("unknown-message-destination")
+ val service = new PekkoActorRefMappingService(newActorService(workerId,
parent))
+ val first = networkMessageTo(destination, messageId = 1L, sequenceNumber =
0L)
+ val second = networkMessageTo(destination, messageId = 2L, sequenceNumber
= 1L)
+
+ service.forwardToActor(first)
+ val lookup = parent.expectMsgType[GetActorRef]
+ assert(lookup.id == destination)
+ assert(lookup.replyTo == Set(service.self))
+
+ service.forwardToActor(second)
+ parent.expectNoMessage(100.millis)
+
+ service.clearQueriedActorRefs()
+ service.forwardToActor(networkMessageTo(destination, messageId = 3L,
sequenceNumber = 2L))
+ assert(parent.expectMsgType[GetActorRef].id == destination)
+ }
+
+ "registerActorRef" should "drain stashed messages in FIFO order and notify
controller waiters" in {
+ val parent = TestProbe()
+ val destination = ActorVirtualIdentity("controller-wait-destination")
+ val waiterOne = TestProbe()
+ val waiterTwo = TestProbe()
+ val registered = TestProbe()
+ val service = new PekkoActorRefMappingService(newActorService(CONTROLLER,
parent))
+ val first = networkMessageTo(destination, messageId = 10L, sequenceNumber
= 0L)
+ val second = networkMessageTo(destination, messageId = 11L, sequenceNumber
= 1L)
+
+ service.forwardToActor(first)
+ service.forwardToActor(second)
+ service.retrieveActorRef(destination, Set(waiterOne.ref, waiterTwo.ref))
+
+ service.registerActorRef(destination, registered.ref)
+
+ registered.expectMsg(first)
+ registered.expectMsg(second)
+ waiterOne.expectMsg(RegisterActorRef(destination, registered.ref))
+ waiterTwo.expectMsg(RegisterActorRef(destination, registered.ref))
+ }
+
+ "retrieveActorRef and removeActorRef" should "reply for known refs and
remove their reverse lookup" in {
+ val parent = TestProbe()
+ val destination = ActorVirtualIdentity("registered-destination")
+ val registered = TestProbe()
+ val waiter = TestProbe()
+ val service = new PekkoActorRefMappingService(newActorService(workerId,
parent))
+
+ service.registerActorRef(destination, registered.ref)
+ service.retrieveActorRef(destination, Set(waiter.ref))
+
+ waiter.expectMsg(RegisterActorRef(destination, registered.ref))
+ assert(service.hasActorRef(destination))
+ assert(service.getActorRef(destination) == registered.ref)
+
assert(service.findActorVirtualIdentity(registered.ref).contains(destination))
+
+ service.removeActorRef(destination)
+
+ assert(!service.hasActorRef(destination))
+ assert(service.findActorVirtualIdentity(registered.ref).isEmpty)
+ }
+
+ "retrieveActorRef" should "swallow a failed parent lookup and still ask
again for the same id" in {
+ val parent = TestProbe()
+ 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, for the first read only.
+ val actorService = new FailingParentActorService(workerId,
newContext(parent))
+ val service = new PekkoActorRefMappingService(actorService)
+ actorService.failuresLeft = 1
+
+ service.retrieveActorRef(destination, Set(waiter.ref))
+
+ // Nothing was asked and nobody was told: the lookup simply did not happen.
+ parent.expectNoMessage(100.millis)
+ waiter.expectNoMessage(100.millis)
+
+ // ...and the id was not recorded as queried, so the next message bound
for it re-asks. Marking
+ // it would strand every message for that destination: the reply that
clears the stash only ever
+ // arrives in response to a `GetActorRef` that was actually sent.
+ service.retrieveActorRef(destination, Set(waiter.ref))
+
+ 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. */
+class ActorRefMappingServiceContextHolder extends Actor {
+ override def receive: Receive = { case _ => () }
+}
+
+/** A [[PekkoActorService]] whose first `failuresLeft` parent lookups throw. */
+class FailingParentActorService(id: ActorVirtualIdentity, actorContext:
ActorContext)
+ extends PekkoActorService(id, actorContext) {
+
+ var failuresLeft: Int = 0
+
+ override def parent: ActorRef = {
+ if (failuresLeft > 0) {
+ failuresLeft -= 1
+ throw new IllegalStateException("parent is unreachable")
+ }
+ super.parent
+ }
+}
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
new file mode 100644
index 0000000000..2beaf4a428
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala
@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+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.{
+ 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 {
+
+ //
---------------------------------------------------------------------------
+ // Fixtures
+ //
---------------------------------------------------------------------------
+
+ private val actorId: ActorVirtualIdentity = ActorVirtualIdentity("worker-1")
+
+ private def emptyQueue(): LinkedBlockingQueue[DPInputQueueElement] =
+ new LinkedBlockingQueue[DPInputQueueElement]()
+
+ private def freshManager: InputManager = new InputManager(actorId,
emptyQueue())
+
+ // A one-field schema is enough to build real Tuple instances for the batch
tests.
+ private val intAttr = new Attribute("v", AttributeType.INTEGER)
+ private val schema: Schema = Schema().add(intAttr)
+ private def tuple(value: Int): Tuple =
+ Tuple.builder(schema).add(intAttr, Integer.valueOf(value)).build()
+
+ private def channelId(from: String, to: String): ChannelIdentity =
+ ChannelIdentity(ActorVirtualIdentity(from), ActorVirtualIdentity(to),
isControl = false)
+
+ //
---------------------------------------------------------------------------
+ // getAllPorts / addPort / getPort
+ //
---------------------------------------------------------------------------
+
+ "InputManager.getAllPorts (fresh)" should "be empty" in {
+ val mgr = freshManager
+ assert(mgr.getAllPorts.isEmpty)
+ }
+
+ "InputManager.addPort" should
+ "register a fresh WorkerPort under the supplied portId (no reader threads
when URI list is empty)" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings =
List.empty)
+ assert(mgr.getAllPorts == Set(portId))
+ val port = mgr.getPort(portId)
+ assert(port.schema == schema)
+ assert(!port.completed)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ it should "be a no-op when called a second time for the same portId (first
call wins)" in {
+ // Pin the "each port can only be added and initialized once" comment in
+ // the production code: the second addPort returns early WITHOUT
+ // overwriting the existing WorkerPort. We prove this by mutating the
+ // first port's `completed` flag, then re-adding, then checking the
+ // mutation survives — if the second call had replaced the WorkerPort,
+ // `completed` would revert to its case-class default `false`.
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings =
List.empty)
+ val firstPort = mgr.getPort(portId)
+ firstPort.completed = true
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings =
List.empty)
+ assert(mgr.getPort(portId) eq firstPort, "second addPort must not replace
the WorkerPort")
+ assert(mgr.getPort(portId).completed, "second addPort must not reset the
existing port's state")
+ }
+
+ it should
+ "throw AssertionError when urisToRead.size does not match
partitionings.size" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ // 1 URI but 0 partitionings — sizes diverge.
+ assertThrows[AssertionError] {
+ mgr.addPort(
+ portId,
+ schema,
+ urisToRead = List(new URI("file:///nowhere")),
+ partitionings = List.empty[Partitioning]
+ )
+ }
+ }
+
+ "InputManager.getPort" should "return the WorkerPort previously registered
by addPort" in {
+ val mgr = freshManager
+ val portId = PortIdentity(7)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings =
List.empty)
+ val a = mgr.getPort(portId)
+ val b = mgr.getPort(portId)
+ assert(a eq b, "getPort must be a pure lookup — no fresh WorkerPort each
call")
+ }
+
+ "InputManager.getInputPortReaderThreads" should
+ "be empty when no port has been added with URIs" in {
+ val mgr = freshManager
+ mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty,
partitionings = List.empty)
+ mgr.addPort(PortIdentity(1), schema, urisToRead = List.empty,
partitionings = List.empty)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ //
---------------------------------------------------------------------------
+ // Batch cursor — initBatch, hasUnfinishedInput, getNextTuple,
getCurrentTuple
+ //
---------------------------------------------------------------------------
+
+ "InputManager.hasUnfinishedInput (fresh)" should "be false when no batch has
been initialized" in {
+ val mgr = freshManager
+ assert(!mgr.hasUnfinishedInput)
+ }
+
+ "InputManager.getCurrentTuple (fresh)" should "return null when no batch has
been initialized" in {
+ val mgr = freshManager
+ assert(mgr.getCurrentTuple == null)
+ }
+
+ "InputManager.initBatch" should
+ "swap in a new batch, reset the cursor, and surface the channel id" in {
+ val mgr = freshManager
+ val cid = channelId("upstream", "worker-1")
+ mgr.initBatch(cid, Array(tuple(10), tuple(20), tuple(30)))
+ assert(mgr.currentChannelId == cid)
+ assert(mgr.hasUnfinishedInput)
+ // Prove the cursor was reset to before-first: the FIRST `getNextTuple`
+ // returns the first element of the batch (idx 0), not idx 1 or later.
+ // If `initBatch` had not reset `currentInputIdx` back to -1, the
+ // first `getNextTuple` would skip past element 0.
+ assert(mgr.getNextTuple == tuple(10))
+ }
+
+ it should "drive getNextTuple to yield elements in order and complete after
the last one" in {
+ val mgr = freshManager
+ val cid = channelId("upstream", "worker-1")
+ val batch = Array(tuple(10), tuple(20), tuple(30))
+ mgr.initBatch(cid, batch)
+ assert(mgr.hasUnfinishedInput)
+ assert(mgr.getNextTuple == tuple(10))
+ assert(mgr.getCurrentTuple == tuple(10))
+ assert(mgr.hasUnfinishedInput)
+ assert(mgr.getNextTuple == tuple(20))
+ assert(mgr.getCurrentTuple == tuple(20))
+ assert(mgr.hasUnfinishedInput)
+ assert(mgr.getNextTuple == tuple(30))
+ assert(mgr.getCurrentTuple == tuple(30))
+ assert(!mgr.hasUnfinishedInput)
+ }
+
+ it should "allow a second initBatch to replace the existing batch and
restart the cursor" in {
+ val mgr = freshManager
+ val cid = channelId("upstream", "worker-1")
+ mgr.initBatch(cid, Array(tuple(1), tuple(2)))
+ assert(mgr.getNextTuple == tuple(1))
+ val cid2 = channelId("upstream-2", "worker-1")
+ mgr.initBatch(cid2, Array(tuple(7), tuple(8), tuple(9)))
+ assert(mgr.currentChannelId == cid2)
+ assert(mgr.hasUnfinishedInput)
+ assert(mgr.getNextTuple == tuple(7))
+ }
+
+ "InputManager.getCurrentTuple (empty batch)" should "return null" in {
+ // Production code path: `inputBatch.isEmpty` → return `null`. Pin the
+ // edge case explicitly so a regression that defaulted to throwing
+ // (or returning a sentinel tuple) would surface.
+ val mgr = freshManager
+ mgr.initBatch(channelId("u", "w"), Array.empty[Tuple])
+ assert(mgr.getCurrentTuple == null)
+ }
+
+ "InputManager.hasUnfinishedInput (empty batch)" should "be false (no
elements remain)" in {
+ val mgr = freshManager
+ mgr.initBatch(channelId("u", "w"), Array.empty[Tuple])
+ assert(!mgr.hasUnfinishedInput)
+ }
+
+ //
---------------------------------------------------------------------------
+ // isPortCompleted — non-materialized path
+ //
---------------------------------------------------------------------------
+ //
+ // For a port with no attached InputPortMaterializationReaderThread,
+ // `isPortCompleted` reads through to `WorkerPort.completed`.
+
+ "InputManager.isPortCompleted (non-materialized)" should
+ "return false for a freshly-added port and true once WorkerPort.completed
is set" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings =
List.empty)
+ assert(!mgr.isPortCompleted(portId))
+ mgr.getPort(portId).completed = true
+ assert(mgr.isPortCompleted(portId))
+ }
+
+ //
---------------------------------------------------------------------------
+ // startInputPortReaderThreads — no-op when nothing is registered
+ //
---------------------------------------------------------------------------
+
+ "InputManager.startInputPortReaderThreads (no reader threads)" should
+ "be a safe no-op" in {
+ val mgr = freshManager
+ mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty,
partitionings = List.empty)
+ 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
new file mode 100644
index 0000000000..2004bf489c
--- /dev/null
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.metadata
+
+import org.eclipse.jgit.api.Git
+import org.eclipse.jgit.api.errors.GitAPIException
+import org.eclipse.jgit.revwalk.RevCommit
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path}
+import java.util.UUID
+import scala.jdk.CollectionConverters._
+import scala.util.Using
+
+/**
+ * `OPVersion` resolves an operator's version from the git history of the file
+ * that defines it, memoizing the answer in a process-wide static map.
+ *
+ * Its git handle is opened once in a static initializer against `TEXERA_HOME`
+ * (defaulting to the working directory). Whether that succeeds depends
entirely
+ * on how the tree was checked out — inside a `git worktree` the `.git` entry
is
+ * a file rather than a directory and jgit raises
`RepositoryNotFoundException`,
+ * leaving the handle null; in a plain clone the handle opens and an unknown
path
+ * yields an empty log instead. Rather than assert whatever both happen to
share,
+ * this spec pins the handle to a known state: every test swaps the private
static
+ * `git` field to a handle it controls (or to null) for its duration and
restores
+ * the original value afterwards, so the assertions describe one
deterministic code
+ * path regardless of how the tree was checked out (and no other suite in the
JVM
+ * is affected).
+ *
+ * The three resolution outcomes are pinned against throwaway repositories
built
+ * in a temp directory, so nothing here depends on the checkout the suite
runs in:
+ *
+ * - 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`), 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 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.
+ *
+ * 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.
+ */
+class OPVersionSpec extends AnyFlatSpec with Matchers {
+
+ /** The cache is static and shared, so every test uses a name nothing else
can collide with. */
+ private def uniqueName(): String = s"OPVersionSpec-${UUID.randomUUID()}"
+
+ private def uniqueMissingPath(): String =
s"no/such/operator/path/${UUID.randomUUID()}"
+
+ private def declaredField(name: String): java.lang.reflect.Field = {
+ val field = classOf[OPVersion].getDeclaredField(name)
+ field.setAccessible(true)
+ field
+ }
+
+ /** The private static memo table, so tests can seed it and clean up after
themselves. */
+ private def opMap: java.util.Map[String, String] = {
+ declaredField("opMap").get(null).asInstanceOf[java.util.Map[String,
String]]
+ }
+
+ /**
+ * Runs `body` with `OPVersion`'s private static git handle forced to
`handle`, so a
+ * chosen resolution path is exercised deterministically, then restores
whatever was
+ * there.
+ */
+ private def withGit[T](handle: Git)(body: => T): T = {
+ val field = declaredField("git")
+ val original = field.get(null)
+ field.set(null, handle)
+ try body
+ finally field.set(null, original)
+ }
+
+ private def withNullGit[T](body: => T): T = withGit(null)(body)
+
+ private def withCleanCache[T](names: String*)(body: => T): T =
+ try body
+ finally names.foreach(opMap.remove)
+
+ private def deleteRecursively(path: Path): Unit =
+ if (Files.exists(path)) {
+ Using.resource(Files.walk(path)) { stream =>
+ stream.iterator().asScala.toSeq.reverse.foreach { p =>
+ // best-effort: jgit can keep Windows handles on .git objects;
leftover temp
+ // files are harmless and get reaped by the OS.
+ try Files.deleteIfExists(p)
+ catch { case _: java.io.IOException => () }
+ }
+ }
+ }
+
+ /** Runs `body` against a throwaway repository, closing the handle and
reaping the directory. */
+ private def withTempRepo[T](prefix: String)(body: (Git, Path) => T): T = {
+ val dir = Files.createTempDirectory(prefix)
+ val handle = Git.init().setDirectory(dir.toFile).call()
+ try body(handle, dir)
+ finally {
+ handle.close()
+ deleteRecursively(dir)
+ }
+ }
+
+ /** Writes `name` and commits just that path; jgit needs an explicit
identity here. */
+ private def commitFile(handle: Git, dir: Path, name: String, content:
String): RevCommit = {
+ Files.write(dir.resolve(name), content.getBytes(StandardCharsets.UTF_8))
+ handle.add().addFilepattern(name).call()
+ handle
+ .commit()
+ .setMessage(s"touch $name")
+ .setAuthor("texera-test", "[email protected]")
+ .setCommitter("texera-test", "[email protected]")
+ .call()
+ }
+
+ // -- resolution against a repository with history
----------------------------
+
+ "OPVersion.getVersion" should "resolve the newest commit that touched the
operator's own path" in {
+ val alphaOp = uniqueName()
+ val betaOp = uniqueName()
+ withCleanCache(alphaOp, betaOp) {
+ withTempRepo("opversion-history") { (handle, dir) =>
+ // Three commits, interleaved so every wrong answer is a different
hash from
+ // the right one: alpha's answer is neither the tip of the log (that
is beta's)
+ // nor the oldest commit that touched it.
+ val alphaFirst = commitFile(handle, dir, "alpha.txt", "alpha v1")
+ val alphaHead = commitFile(handle, dir, "alpha.txt", "alpha v2")
+ val betaHead = commitFile(handle, dir, "beta.txt", "beta v1")
+
+ // Guard the fixture: three distinct hashes, so none of the assertions
below
+ // can pass by coincidence.
+ Set(alphaFirst.getName, alphaHead.getName, betaHead.getName) should
have size 3
+
+ withGit(handle) {
+ OPVersion.getVersion(alphaOp, "alpha.txt") shouldBe alphaHead.getName
+ OPVersion.getVersion(betaOp, "beta.txt") shouldBe betaHead.getName
+ }
+ }
+ }
+ }
+
+ it should "memoize the resolved commit hash under the operator name" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withTempRepo("opversion-memo") { (handle, dir) =>
+ val head = commitFile(handle, dir, "alpha.txt", "alpha v1")
+
+ withGit(handle) {
+ opMap.containsKey(name) shouldBe false
+ OPVersion.getVersion(name, "alpha.txt") shouldBe head.getName
+ opMap.get(name) shouldBe head.getName
+ }
+ }
+ }
+ }
+
+ it should "swallow a git failure instead of propagating it to the caller" in
{
+ val name = uniqueName()
+ withCleanCache(name) {
+ withTempRepo("opversion-unborn") { (handle, _) =>
+ // A freshly initialised repository has an unborn HEAD, so
LogCommand.call()
+ // raises NoHeadException -- a checked GitAPIException, a different
catch from
+ // the NullPointerException that a missing handle produces. Asserted
first so
+ // this test cannot pass vacuously if jgit ever starts returning an
empty log.
+ a[GitAPIException] should be thrownBy
+ handle.log().addPath("alpha.txt").setMaxCount(1).call()
+
+ withGit(handle) {
+ noException should be thrownBy OPVersion.getVersion(name,
"alpha.txt")
+ }
+ }
+ }
+ }
+
+ 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 {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withNullGit {
+ OPVersion.getVersion(name, uniqueMissingPath()) shouldBe "N/A"
+ }
+ }
+ }
+
+ it should "never return null, whatever it resolves" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withNullGit {
+ OPVersion.getVersion(name, "common/workflow-operator/src/main/scala")
should not be null
+ }
+ }
+ }
+
+ it should "memoize the resolved version under the operator name" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withNullGit {
+ opMap.containsKey(name) shouldBe false
+ val resolved = OPVersion.getVersion(name, uniqueMissingPath())
+ opMap.containsKey(name) shouldBe true
+ opMap.get(name) shouldBe resolved
+ }
+ }
+ }
+
+ // -- memoization contract
----------------------------------------------------
+
+ it should "serve the memoized value and ignore the path on subsequent calls"
in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withNullGit {
+ // Seed a value no lookup could ever produce: if the cache were
bypassed,
+ // the call below would recompute and answer "N/A" instead.
+ opMap.put(name, "seeded-version")
+ OPVersion.getVersion(name, uniqueMissingPath()) shouldBe
"seeded-version"
+ OPVersion.getVersion(name, "a/completely/different/path") shouldBe
"seeded-version"
+ }
+ }
+ }
+
+ it should "key the cache by operator name rather than by path" in {
+ val first = uniqueName()
+ val second = uniqueName()
+ withCleanCache(first, second) {
+ withNullGit {
+ opMap.put(first, "version-of-first")
+ opMap.put(second, "version-of-second")
+
+ val sharedPath = "common/workflow-operator/src/main/scala"
+ OPVersion.getVersion(first, sharedPath) shouldBe "version-of-first"
+ OPVersion.getVersion(second, sharedPath) shouldBe "version-of-second"
+ }
+ }
+ }
+
+ it should "resolve unknown paths consistently across repeated calls" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withNullGit {
+ val path = uniqueMissingPath()
+ val first = OPVersion.getVersion(name, path)
+ val second = OPVersion.getVersion(name, path)
+ first shouldBe "N/A"
+ second shouldBe first
+ }
+ }
+ }
+}
