You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/23 04:35:56 UTC
(texera) branch gh-readonly-queue/release/v1.2/pr-7809-41cc7c5196b88ffba323b4b0a2b458d825893720 created (now 203286c103)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/release/v1.2/pr-7809-41cc7c5196b88ffba323b4b0a2b458d825893720
in repository https://gitbox.apache.org/repos/asf/texera.git
at 203286c103 fix(amber, operator, v1.2): keep error information on three failure paths (#7809)
This branch includes the following new commits:
new 203286c103 fix(amber, operator, v1.2): keep error information on three failure paths (#7809)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(amber, operator, v1.2): keep error information on three failure paths (#7809)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/release/v1.2/pr-7809-41cc7c5196b88ffba323b4b0a2b458d825893720
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 203286c1039e4d2e8c88c14883f28da5730ab3d3
Author: github-actions[bot] <41...@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 <xi...@uci.edu>
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
---
.../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", "texera-test@example.com")
+ .setCommitter("texera-test", "texera-test@example.com")
+ .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
+ }
+ }
+ }
+}