This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8938-6a29c59c6312da95826b269fca2a6a14006fb34a in repository https://gitbox.apache.org/repos/asf/texera.git
commit 0ee5bff99f795451153d3c1fe61da22825b40b13 Author: Xinyuan Lin <[email protected]> AuthorDate: Wed Oct 14 12:46:44 2026 +0000 refactor(amber): count a materialized input port as completed once its END is processed (#8938) ### What changes were proposed in this PR? `InputManager.isPortCompleted` now treats a port read from materialization the same as a port fed by a link: it counts as completed once END_CHANNEL has been processed on it. `EndChannelHandler` marks that by setting `WorkerPort.completed`. Before, such a port reported its reader thread's `finished` flag. The thread sets that flag only after enqueuing END. So a DP thread waiting on the queue, typically the worker that got no tuples, could run `EndChannelHandler` before the flag was set. It then saw the port as not completed and skipped `finalizeOutput`. Nothing called it again, so the worker never finished and its region hung until the test deadline. That is the intermittent `TimeoutException: 1.minutes+30.seconds` in `LoopIntegrationSpec`. ``` Before: reader enqueues END -> DP processes END, flag not set yet -> output never finalized -> region hangs After: reader enqueues END -> DP processes END, port completed -> output finalized -> region completes ``` Two related effects of the change: - The flag could also be set while END was still queued. A worker with a second materialized input no longer finalizes before that port's END has been processed. - `startInputPortReaderThreads` uses the same check to skip ports that already completed. The second `StartWorker` of a two-phase region arrives only after the coordinator has the port's `PortCompleted`, which the worker sends after processing END. So it skips the same ports as before. - Nothing reads the reader thread's `finished` flag any more, so it is removed, and it can't stand in for port completion again. The reader-thread tests that read it now check the END_CHANNEL the reader enqueues instead. ### Any related issues, documentation, discussions? Closes #8937. The two `LoopIntegrationSpec` cases added in #8639 hit the race most often; the race itself dates from #3460. ### How was this PR tested? Tests first: the first commit adds the tests and fails on `main`, and the second commit makes them pass. - `InputManagerSpec`: a materialized port is completed once END has been processed, even before its reader thread has returned. It is not completed while END is still queued, even after the thread has returned. - `EndChannelHandlerSpec`: when END arrives on a materialized input port before its reader thread has returned, the worker emits the output `FinalizePort` and `FinalizeExecutor`. On `main` it emits only the input port's `FinalizePort`, which is the hang. ``` sbt "WorkflowExecutionService/testOnly *InputManagerSpec *EndChannelHandlerSpec *StartHandlerSpec *DataProcessorSpec *EndHandlerSpec *StartChannelHandlerSpec *AssignPortHandlerSpec *DPThreadSpec *WorkerSpec" sbt "WorkflowExecutionService/scalafmtCheckAll" "WorkflowExecutionService/scalafixAll --check" ``` `InputPortMaterializationReaderThreadSpec` reads real Iceberg tables. Locally only its two cases that don't need storage ran; CI runs the rest. `LoopIntegrationSpec`, run locally on Windows with the sbt JVM and its Python workers pinned to 3 CPUs (affinity mask `0x7`), one case per JVM: | Case | `main` | This PR | |---|---|---| | bind a `$i` reference after two branches of the loop body meet again | 3 of 5 runs hung on the first attempt; 1 also hung on the retry and failed as on CI | 10 of 10 passed on the first attempt, 42–44 s each | | bind the inner loop's variable where it shadows the outer loop's | (fails on CI, see #8937) | 10 of 10 passed on the first attempt, 64–70 s each | Unpinned, the whole spec passes with this PR: 8 of 8 cases, 29–59 s each, no retries. ``` AMBER_TEST_FILTER= STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec -- -z \"after two branches of the loop body meet again\"" AMBER_TEST_FILTER=integration-only STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec" ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- .../architecture/messaginglayer/InputManager.scala | 17 ++------ .../InputPortMaterializationReaderThread.scala | 9 +--- .../messaginglayer/InputManagerSpec.scala | 45 +++++++++++++++++++ .../InputPortMaterializationReaderThreadSpec.scala | 45 ++++++++----------- .../promisehandlers/EndChannelHandlerSpec.scala | 50 ++++++++++++++++++---- 5 files changed, 110 insertions(+), 56 deletions(-) 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 c017eb238b..2a8c6b4bbc 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 @@ -116,20 +116,11 @@ class InputManager( def getPort(portId: PortIdentity): WorkerPort = ports(portId) /** - * For ports that read from materialization, the port completion is marked by the finish of the reader thread. - * For other ports that connect to upstream links, the completion is marked by the completion the port. + * Whether END_CHANNEL has been processed on every channel of the port (`EndChannelHandler` then + * marks it completed), for a port read from materialization as well: what its reader thread has + * done says nothing about whether the DP thread has processed the END it enqueued. */ - def isPortCompleted(portId: PortIdentity): Boolean = { - if ( - !this.inputPortMaterializationReaderThreads - .contains(portId) || this.inputPortMaterializationReaderThreads(portId).isEmpty - ) { - this.getPort(portId).completed - } else { - val existingThread = this.inputPortMaterializationReaderThreads(portId).head - existingThread.finished - } - } + def isPortCompleted(portId: PortIdentity): Boolean = this.getPort(portId).completed def hasUnfinishedInput: Boolean = inputBatch != null && currentInputIdx + 1 < inputBatch.length diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThread.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThread.scala index f1ece04b5a..f090c4c803 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThread.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThread.scala @@ -55,7 +55,7 @@ import org.apache.texera.amber.util.VirtualIdentityUtils.getFromActorIdForInputP import java.net.URI import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong} +import java.util.concurrent.atomic.AtomicLong import scala.collection.mutable.ArrayBuffer class InputPortMaterializationReaderThread( @@ -75,12 +75,6 @@ class InputPortMaterializationReaderThread( } private val partitioner = toPartitioner(partitioning, workerActorId) private val batchSize = ApplicationConfig.defaultDataTransferBatchSize - private val isFinished = new AtomicBoolean(false) - - /** - * Whether the reader thread has completed. - */ - def finished: Boolean = isFinished.get() /** * Read from the materialization storage, and mimics the behavior of an upstream worker's output manager. @@ -141,7 +135,6 @@ class InputPortMaterializationReaderThread( // Flush any remaining tuples in the buffer. if (buffer.nonEmpty) flush() emitECM(METHOD_END_CHANNEL, PORT_ALIGNMENT) - isFinished.set(true) } catch { case e: Exception => throw new RuntimeException(s"Error reading input port materializations: ${e.getMessage}", e) 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 2beaf4a428..477494f9b5 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 @@ -215,6 +215,49 @@ class InputManagerSpec extends AnyFlatSpec { assert(mgr.isPortCompleted(portId)) } + // --------------------------------------------------------------------------- + // isPortCompleted — materialized path + // --------------------------------------------------------------------------- + // + // A port read from materialization is fed by a reader thread that enqueues END_CHANNEL. The DP + // thread may process that END before the thread returns, and the thread may return while END is + // still queued, so the port is completed when END has been processed (`WorkerPort.completed`, + // set by EndChannelHandler), whatever state the reader thread is in. + + private val materializedUri = new URI("file:///nowhere") + + private def broadcastToSelf: Partitioning = + BroadcastPartitioning(batchSize = 1, channels = Seq(channelId("upstream", "worker-1"))) + + "InputManager.isPortCompleted (materialized)" should + "be true once END has been processed, before the reader thread has returned" in { + val mgr = freshManager + val portId = PortIdentity(0) + mgr.addPort(portId, schema, List(materializedUri), List(broadcastToSelf)) + val reader = mgr.getInputPortReaderThreads(portId).head + assert(!mgr.isPortCompleted(portId)) + + // What EndChannelHandler does on END, while the reader thread has not even started. + mgr.getPort(portId).completed = true + + assert(reader.getState == Thread.State.NEW) + assert(mgr.isPortCompleted(portId)) + } + + it should "be false while END is still queued, even after the reader thread has returned" in { + val mgr = freshManager + val portId = PortIdentity(0) + mgr.addPort(portId, schema, List.empty, List.empty) + val returnedReader = new NoOpReaderThread + returnedReader.start() + returnedReader.join() + installReaderThreads(mgr, portId, List(returnedReader)) + + // The reader enqueued END and returned, but the DP thread has not processed END yet. + assert(returnedReader.getState == Thread.State.TERMINATED) + assert(!mgr.isPortCompleted(portId)) + } + // --------------------------------------------------------------------------- // startInputPortReaderThreads — no-op when nothing is registered // --------------------------------------------------------------------------- @@ -276,6 +319,8 @@ class InputManagerSpec extends AnyFlatSpec { val alreadyStarted = new NoOpReaderThread alreadyStarted.start() alreadyStarted.join() // now TERMINATED, so a second start() is guaranteed to fail + // Registered first, as addPort always does before building a port's reader threads. + mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty, partitionings = List.empty) installReaderThreads(mgr, PortIdentity(0), List(alreadyStarted)) val thrown = intercept[RuntimeException] { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThreadSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThreadSpec.scala index d64bb31c3b..9651b2d35a 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThreadSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThreadSpec.scala @@ -157,20 +157,15 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { hashAttributeNames = Seq("id") ) - /** - * Runs the reader inline and returns everything it enqueued, in order, together with the reader - * itself so its completion flag can be read. - */ + /** Runs the reader inline and returns everything it enqueued, in order. */ private def replay( uri: URI, partitioning: Partitioning, readerFor: ActorVirtualIdentity = workerId - ): (InputPortMaterializationReaderThread, Seq[WorkflowFIFOMessage]) = { + ): Seq[WorkflowFIFOMessage] = { val queue = new LinkedBlockingQueue[DPInputQueueElement]() - val reader = - new InputPortMaterializationReaderThread(uri, queue, readerFor, partitioning) - reader.run() - (reader, drain(queue)) + new InputPortMaterializationReaderThread(uri, queue, readerFor, partitioning).run() + drain(queue) } private def drain(queue: LinkedBlockingQueue[DPInputQueueElement]): Seq[WorkflowFIFOMessage] = { @@ -197,9 +192,9 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { behavior of "InputPortMaterializationReaderThread" - it should "bracket the replay with START_CHANNEL and END_CHANNEL and mark itself finished" in { + it should "bracket the replay with START_CHANNEL and END_CHANNEL" in { val uri = materialize("bracket", tuples = Seq(tuple(1))) - val (reader, messages) = replay(uri, broadcastTo(workerId)) + val messages = replay(uri, broadcastTo(workerId)) // The markers are the first and last things on the channel: a downstream input port that saw // data before START_CHANNEL, or END_CHANNEL before the last batch, would mis-order the port. @@ -218,26 +213,24 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { } assert(start.ecmType == NO_ALIGNMENT) assert(end.ecmType == PORT_ALIGNMENT) - assert(reader.finished) } - it should "not report itself finished before the replay runs" in { + it should "enqueue nothing before the replay runs" in { val queue = new LinkedBlockingQueue[DPInputQueueElement]() - val reader = new InputPortMaterializationReaderThread( + new InputPortMaterializationReaderThread( baseURI("unstarted"), queue, workerId, broadcastTo(workerId) ) - assert(!reader.finished) assert(queue.isEmpty) } it should "send every message on one channel from a URI-derived sender, numbered from zero" in { val uri = materialize("channel", tuples = Seq(tuple(1)), states = Seq((State(Map("k" -> 1)), 0L, ""))) - val (_, messages) = replay(uri, broadcastTo(workerId)) + val messages = replay(uri, broadcastTo(workerId)) // The dummy sender is derived from the URI *and* the receiving worker, so two workers reading // the same materialization do not share a channel (and neither collides with a real upstream). @@ -260,7 +253,7 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { ) val uri = materialize("states", tuples = Seq(tuple(1), tuple(2)), states = states) - val (_, messages) = replay(uri, broadcastTo(workerId)) + val messages = replay(uri, broadcastTo(workerId)) // Downstream operators need their state set up before the tuples arrive, so every state // precedes every tuple regardless of how the two tables were written. @@ -277,7 +270,7 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { it should "emit no state frame when the port materialized no state" in { val uri = materialize("nostate", tuples = Seq(tuple(1))) - val (_, messages) = replay(uri, broadcastTo(workerId)) + val messages = replay(uri, broadcastTo(workerId)) assert(stateFrames(messages).isEmpty) assert(deliveredIds(messages) == Seq(1)) @@ -287,7 +280,7 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { val ids = 1 to 5 val uri = materialize("broadcast", tuples = ids.map(tuple)) - val (_, messages) = replay(uri, broadcastTo(workerId, siblingWorkerId)) + val messages = replay(uri, broadcastTo(workerId, siblingWorkerId)) // Under the batch size, everything leaves in the single trailing flush. assert(dataFrames(messages).size == 1) @@ -299,7 +292,7 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { // Broadcast to the sibling only: `allReceivers` never contains this worker, so the membership // test fails for every tuple. - val (reader, messages) = replay(uri, broadcastTo(siblingWorkerId)) + val messages = replay(uri, broadcastTo(siblingWorkerId)) assert(deliveredIds(messages).isEmpty) assert(dataFrames(messages).isEmpty) @@ -310,7 +303,6 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { EmbeddedControlMessageIdentity("EndChannel") ) ) - assert(reader.finished) } it should "split the materialization between the workers of a hash-shuffled link" in { @@ -318,8 +310,8 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { val uri = materialize("shuffle", tuples = ids.map(tuple)) val partitioning = hashShuffleTo(workerId, siblingWorkerId) - val (_, mine) = replay(uri, partitioning) - val (_, siblings) = replay(uri, partitioning, readerFor = siblingWorkerId) + val mine = replay(uri, partitioning) + val siblings = replay(uri, partitioning, readerFor = siblingWorkerId) // Each worker's reader takes exactly its own buckets: together they cover the materialization // once. A filter that ignored the partitioner would duplicate every tuple; one that dropped @@ -337,7 +329,7 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { val ids = 1 to (batchSize + 1) val uri = materialize("batching", tuples = ids.map(tuple)) - val (_, messages) = replay(uri, broadcastTo(workerId)) + val messages = replay(uri, broadcastTo(workerId)) assert(dataFrames(messages).map(_.frame.length) == Seq(batchSize, 1)) assert(deliveredIds(messages) == ids) @@ -358,9 +350,8 @@ class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { assert(failure.getMessage.startsWith("Error reading input port materializations: ")) assert(failure.getCause != null) assert(failure.getMessage.endsWith(failure.getCause.getMessage)) - // START_CHANNEL is emitted before the try, so the marker is already on the queue... + // START_CHANNEL is emitted before the try, so the marker is already on the queue, but + // END_CHANNEL never is: the port is never completed. assert(ecmIds(drain(queue)) == Seq(EmbeddedControlMessageIdentity("StartChannel"))) - // ...but the channel never completes, which is what stops the port being called complete. - assert(!reader.finished) } } diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala index d768b93c7d..67950bb2bf 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala @@ -44,7 +44,10 @@ import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ EmptyRequest } import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn -import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.OneToOnePartitioning +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + OneToOnePartitioning +} import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ DPInputQueueElement, MainThreadDelegateMessage @@ -60,6 +63,7 @@ import org.apache.texera.amber.engine.common.ambermessage.{StateFrame, WorkflowF import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.scalatest.flatspec.AnyFlatSpec +import java.net.URI import java.util.concurrent.LinkedBlockingQueue import scala.collection.mutable.ArrayBuffer import scala.util.control.ControlThrowable @@ -88,11 +92,10 @@ import scala.util.control.ControlThrowable * The `forall` in step 5 is why the fixture always registers two input ports: a worker whose * other input port is still open must NOT finalize its output, or the downstream region would * see the executor finish while data is still arriving on the second port. The fixture pins that - * `forall`, but only over link-fed ports: `InputManager.isPortCompleted` branches, and for a port - * fed from materialization it reports the reader THREAD's `finished` flag and ignores - * `completed` entirely. That arm needs storage-backed input ports, which this fixture cannot - * build without real Iceberg documents, so substituting `getPort(p).completed` for - * `isPortCompleted(p)` is not distinguishable here — a known, stated gap rather than coverage. + * `forall`. A port fed from materialization is completed the same way, by the END this handler + * processes, whatever its reader thread has done since enqueuing that END, so the fixture can + * register such a port with a reader thread that never starts and still expect the output to be + * finalized. * * `produceStateOnFinish` has no override in main today — `OperatorExecutor` returns `None` for * every port — so the "operator produced a state" arm is reachable only through a test executor. @@ -140,10 +143,14 @@ class EndChannelHandlerSpec extends AnyFlatSpec { * @param otherInputPortCompleted whether the *other* input port has already finished. The * common single-input case behaves like `true`: the channel * being ended is the last one open. + * @param currentPortMaterialized whether the ended port is read from materialization, by a + * reader thread that is never started here, so it has not + * returned when END is processed. */ private class Fixture( executor: OperatorExecutor, - otherInputPortCompleted: Boolean = true + otherInputPortCompleted: Boolean = true, + currentPortMaterialized: Boolean = false ) { val sent: ArrayBuffer[WorkflowFIFOMessage] = ArrayBuffer() @@ -158,7 +165,16 @@ class EndChannelHandlerSpec extends AnyFlatSpec { // Two input ports on two different channels; only one of them is the channel the ECM came in // on, so a handler that picked an arbitrary channel would read the wrong port. - dp.inputManager.addPort(currentPortId, schema, List.empty, List.empty) + if (currentPortMaterialized) { + dp.inputManager.addPort( + currentPortId, + schema, + List(new URI("file:///end-channel-spec")), + List(BroadcastPartitioning(batchSize, Seq(currentChannelId))) + ) + } else { + dp.inputManager.addPort(currentPortId, schema, List.empty, List.empty) + } dp.inputManager.addPort(otherPortId, schema, List.empty, List.empty) dp.inputGateway.getChannel(currentChannelId).setPortId(currentPortId) dp.inputGateway.getChannel(otherChannelId).setPortId(otherPortId) @@ -427,6 +443,24 @@ class EndChannelHandlerSpec extends AnyFlatSpec { assert(executor.calls.map(c => (c._1, c._2)) == Seq(("state", 3), ("finish", 3))) } + it should "finalize the output when the last input port is read from materialization, before its reader thread has returned" in { + val fixture = new Fixture(new RecordingExecutor(), currentPortMaterialized = true) + val reader = fixture.dp.inputManager.getInputPortReaderThreads(currentPortId).head + + fixture.endChannel() + + // The DP thread can process END before the reader thread that enqueued it returns. Leaving + // the output unfinalized then would hang the region: nothing ends it later. + assert(reader.getState == Thread.State.NEW) + assert( + fixture.drainOutput() == List( + FinalizePort(currentPortId, input = true), + FinalizePort(outputPortId, input = false), + FinalizeExecutor() + ) + ) + } + it should "finalize the output once the last input port completes" in { val fixture = new Fixture(new RecordingExecutor(), otherInputPortCompleted = true)
