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-7729-16ba619bdf4d0037299c6de772f414c27d043ffe in repository https://gitbox.apache.org/repos/asf/texera.git
commit 7fd4f892766f0a01c2d0c7a99cebc6c8d2ab7037 Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Aug 17 07:07:38 2026 +0000 test(amber): cover the materialization reader thread and the executor initializer (#7729) ### What changes were proposed in this PR? Two worker-side files had no spec. Adds one each, 14 tests total. Measured by running the full amber unit suite under the CI environment (`STORAGE_ICEBERG_CATALOG_TYPE=postgres`, `TEXERA_SERVICE_LOG_LEVEL=WARN`) before and after, so these are directly comparable to the reported figures — the baseline reproduced all of them exactly: | File | Before | After | |---|---|---| | `InputPortMaterializationReaderThread.scala` | 52/63 (82.5%), 7 missed + 4 partial | **62/63 (98.4%)**, 0 missed | | `DataProcessorRPCHandlerInitializer.scala` | 5/15 (33.3%), 8 missed + 2 partial | **8/15 (53.3%)** | Instruction coverage: reader thread 85.1% -> **100.0%**; initializer 81.0% -> 89.1%. Run on their own, the two specs take the reader thread to 63/63 lines and 343/343 instructions. **This is +13 lines, and I would rather state that plainly than dress it up.** What earns the PR is the pinning rather than the count: the reader thread's replay ordering, batching, partition filtering and failure wrapping were all unverified, and eight distinct mutants now die on them. ### A third target was assessed and rejected `ClientActor.scala` (65.2%) is **saturated, not undertested**, and roughly 190 lines of test bought exactly 1 line, 1 branch and 3 instructions — so that work was dropped rather than shipped. Its 7 remaining misses are the companion object's own `<init>`, four `case class` declaration lines carrying 112 of the file's 176 branches inside scalac-generated `equals`/`hashCode`/`productElement` (all with zero covered branches), and two structural lines. The cause is that the e2e specs are not tagged `@IntegrationTest`, so they run in the coverage job and already drive a real Coordinator through a real `ClientActor`. ### Verification 12 mutations, **11 killed, 1 recorded equivalent**, each applied one at a time with the production file diffed after every revert and the failing test read by name. | Mutation | Killed by | |---|---| | exchange `workerIdx` and `workerCount` | builds the named class from the descriptor, worker index and worker count | | exchange `OpExecWithCode`'s code and language fields | sends the code to the Java compiler, surfacing its diagnostics | | install an executor only when none is installed | replaces an executor that was already installed | | empty init info returns `null` instead of throwing | rejects an empty `OpExecInitInfo` instead of installing a null executor | | exchange START and END channel alignments | brackets the replay and marks itself finished | | exchange channel sender and receiver | sends every message from a URI-derived sender, numbered from zero | | `buffer.size >= batchSize` -> `>` | cuts a full batch as soon as the buffer reaches the transfer batch size | | drop the receiver check in the partitioner filter | drops tuples routed to another worker; splits a hash-shuffled link | | drop the cause when wrapping a storage failure | wraps a storage failure in a `RuntimeException` that keeps the cause | | `isFinished.set(true)` -> `false` | two tests | | shift every replayed loop counter by one | replays states ahead of tuples, carrying each row's loop envelope | **The equivalent mutant:** flipping the trailing-flush guard survives. `if (buffer.nonEmpty) flush()` is immediately followed by `emitECM(METHOD_END_CHANNEL, ...)`, whose first statement is `flush()`, and `flush()` itself returns early on an empty buffer. Both legs enqueue the same `DataFrame`, in the same position, with the same sequence number. No test can distinguish them — the guard is dead weight. **One process failure worth recording**, because it is the kind that silently corrupts a mutation table: the partition-filter mutation was first spelled `.nonEmpty`, which collided with a nearby `buffer.nonEmpty`. The revert's exactly-once assertion correctly refused to fire and left the mutant applied, contaminating the four runs after it — their tables showed the earlier mutation's tests failing alongside the real kill. The `git diff` guard caught it; the file was restored, the mutation respelled as `.exists(_ => true)`, and those runs redone. The table above is the clean run. ### Deliberately not included - `newExecFromJavaCode` is uncoverable from a unit test: `JavaRuntimeCompilation.compileCode` passes null options to `compiler.getTask(...)`, so system javac uses `java.class.path`, which under sbt (`fork := false`) does not contain workflow-core. A UDF implementing `OperatorExecutor` cannot compile, so the method never returns normally. - `noOperation` is dead on the Scala path — `NoOperation` is produced and consumed only by the Python worker. The other three `???` members have live coordinator-side call sites but resolve to `???` on a Scala worker; testing them would cement "unimplemented" as a contract. - Remaining partials are the unreachable kinds: a `lazy val` bitmap and the `MatchError` fall-through of a scalapb sealed oneof. A possible defect is reported rather than pinned: `InputManager.startInputPortReaderThreads` wraps a start failure as `new RuntimeException(s"Error starting input port materialization reader thread: ${e.getMessage}")`, **dropping the cause** — unlike the reader thread itself, which passes `e`. That discards the stack trace of the real failure. No production file is touched. ### Any related issues, documentation, discussions? Closes #7728 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.worker.DataProcessorRPCHandlerInitializerSpec org.apache.texera.amber.engine.architecture.worker.managers.InputPortMaterializationReaderThreadSpec" ``` ``` [info] Total number of tests run: 14 [info] Tests: succeeded 14, failed 0, canceled 0, ignored 0, pending 0 ``` The full amber unit suite is green alongside them (1930 passed). `Test/scalafmtCheck` and `Test/scalafix --check` both pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../DataProcessorRPCHandlerInitializerSpec.scala | 147 +++++++++ .../InputPortMaterializationReaderThreadSpec.scala | 366 +++++++++++++++++++++ 2 files changed, 513 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializerSpec.scala new file mode 100644 index 0000000000..18c41d5294 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializerSpec.scala @@ -0,0 +1,147 @@ +/* + * 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.worker + +import org.apache.texera.amber.core.executor.{ + OpExecInitInfo, + OperatorExecutor, + OpExecWithClassName, + OpExecWithCode +} +import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ + DPInputQueueElement, + MainThreadDelegateMessage +} +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.scalatest.flatspec.AnyFlatSpec + +import java.util.concurrent.LinkedBlockingQueue + +/** + * `DataProcessorRPCHandlerInitializer` is mostly a mixin point — it stacks every worker-side + * promise handler onto one object and exposes `dp` to them. Its own logic is `setupExecutor`, + * the shared body behind both `initializeExecutor` (first construction) and `updateExecutor` + * (reconfiguration): it picks a construction strategy from the `OpExecInitInfo` variant and + * installs the result on the `DataProcessor`. + * + * The tests below pin which variant maps to which strategy and that the surrounding arguments + * (`descString`, `workerIdx`, `workerCount`) reach it — `ExecFactory`'s own constructor-resolution + * rules are already pinned by `CoreExecutorReflectionSpec` and are not re-asserted here. + * + * Deliberately NOT covered, and why: `debugCommand`, `evaluatePythonExpression`, + * `retryCurrentTuple` and `noOperation` are all `???`. Asserting that they raise + * `NotImplementedError` would only restate a Scala standard-library default and would cement + * "unimplemented" as an intended contract. `noOperation` is additionally dead — no call site + * exists anywhere in the repository. + */ +class DataProcessorRPCHandlerInitializerSpec extends AnyFlatSpec { + + import DataProcessorRPCHandlerInitializerSpec._ + + private val workerId = ActorVirtualIdentity("Worker:WF1-dp-rpc-init-main-0") + + private def newInitializer(): DataProcessorRPCHandlerInitializer = { + val outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit = _ => () + val dp = + new DataProcessor(workerId, outputHandler, new LinkedBlockingQueue[DPInputQueueElement]()) + new DataProcessorRPCHandlerInitializer(dp) + } + + behavior of "DataProcessorRPCHandlerInitializer.setupExecutor" + + it should "build the named class from the descriptor, worker index and worker count" in { + val initializer = newInitializer() + + // Three distinct values, so a swap or a dropped argument cannot pass: the index is not the + // count, and neither is the descriptor. + initializer.setupExecutor( + OpExecWithClassName(classOf[DescribedExec].getName, "desc-for-worker"), + workerIdx = 3, + workerCount = 7 + ) + + val executor = initializer.dp.executor.asInstanceOf[DescribedExec] + assert(executor.desc == "desc-for-worker") + assert(executor.idx == 3) + assert(executor.workerCount == 7) + } + + it should "replace an executor that was already installed" in { + // `updateExecutor` reconfigures a running worker through this same method, so the assignment + // has to overwrite rather than only initialize. + val initializer = newInitializer() + initializer.setupExecutor(OpExecWithClassName(classOf[DescribedExec].getName, "first"), 0, 1) + val first = initializer.dp.executor + + initializer.setupExecutor(OpExecWithClassName(classOf[DescribedExec].getName, "second"), 0, 1) + + assert(initializer.dp.executor ne first) + assert(initializer.dp.executor.asInstanceOf[DescribedExec].desc == "second") + } + + it should "send OpExecWithCode's code to the Java compiler, surfacing its diagnostics" in { + // The code variant is the Java-UDF path: the string is compiled at runtime rather than looked + // up as a class name. javac here runs on its own classpath and cannot see project classes + // (see CoreExecutorReflectionSpec), so a compiling UDF is not expressible in a test; what is + // observable — and what distinguishes this arm from the class-name arm — is that the code + // itself reaches the compiler, which is why the assertion is on javac's own diagnostic. + val initializer = newInitializer() + + val failure = intercept[RuntimeException] { + initializer.setupExecutor( + OpExecWithCode("public class JavaUDFOpExec { int broken = \"not an int\"; }", "java"), + workerIdx = 0, + workerCount = 1 + ) + } + + assert(failure.getMessage.contains("Error at line")) + assert(failure.getMessage.contains("incompatible types")) + assert(initializer.dp.executor == null, "a failed compilation must not install an executor") + } + + it should "reject an empty OpExecInitInfo instead of installing a null executor" in { + // `OpExecInitInfo` is a protobuf sealed oneof, so "no variant set" is a representable value + // that arrives from the wire; it is rejected loudly rather than left to NPE later. + val initializer = newInitializer() + + val failure = intercept[IllegalArgumentException] { + initializer.setupExecutor(OpExecInitInfo.Empty, workerIdx = 0, workerCount = 1) + } + + assert(failure.getMessage == "Empty executor initialization info") + assert(initializer.dp.executor == null) + } +} + +private object DataProcessorRPCHandlerInitializerSpec { + + /** + * Records everything `newExecFromJavaClassName` is asked to pass through. Public, and with the + * `(String, Int, Int)` constructor that branch resolves to, so reflection inside `ExecFactory` + * can reach it. + */ + class DescribedExec(val desc: String, val idx: Int, val workerCount: Int) + extends OperatorExecutor { + override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = Iterator.empty + } +} 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 new file mode 100644 index 0000000000..d64bb31c3b --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThreadSpec.scala @@ -0,0 +1,366 @@ +/* + * 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.worker.managers + +import org.apache.texera.common.config.ApplicationConfig +import org.apache.texera.amber.core.state.State +import org.apache.texera.amber.core.storage.model.BufferedItemWriter +import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory} +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ChannelIdentity, + EmbeddedControlMessageIdentity, + ExecutionIdentity, + OperatorIdentity, + PhysicalOpIdentity, + WorkflowIdentity +} +import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmbeddedControlMessageType.{ + NO_ALIGNMENT, + PORT_ALIGNMENT +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmbeddedControlMessage +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + HashBasedShufflePartitioning, + Partitioning +} +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ + DPInputQueueElement, + FIFOMessageElement +} +import org.apache.texera.amber.engine.common.ambermessage.{ + DataFrame, + StateFrame, + WorkflowFIFOMessage +} +import org.apache.texera.amber.util.VirtualIdentityUtils.getFromActorIdForInputPortStorage +import org.scalatest.flatspec.AnyFlatSpec + +import java.net.URI +import java.util.concurrent.LinkedBlockingQueue +import scala.collection.mutable.ArrayBuffer + +/** + * `InputPortMaterializationReaderThread` replays a materialized upstream output into a worker's + * input queue, standing in for the upstream worker's output manager: it brackets the replay with + * the same START_CHANNEL / END_CHANNEL embedded control messages a live upstream would send, and + * it applies the upstream link's partitioning itself, because the materialization holds the whole + * port's output and only the slice addressed to this worker may be delivered. + * + * The two documents it reads are real Iceberg tables written through `DocumentFactory` on the + * ambient catalog, the same way `SyncExecutionResourceSpec` / `ExecutionResultServiceSpec` set up + * their result tables — there is no seam to inject storage through, and a fixture that stubbed the + * documents would not exercise `run` at all. Each test hangs off its own operator id, because an + * Iceberg table's identity derives from its URI path and sbt runs amber's suites in one unforked + * JVM. + * + * `run()` is invoked directly rather than through `start()`: the production caller + * (`InputManager.startInputPortReaderThreads`) starts a real thread, but the body is what is under + * test and running it inline makes the queue contents observable without a join/poll race. + */ +class InputPortMaterializationReaderThreadSpec extends AnyFlatSpec { + + /** The worker whose input queue is being filled — i.e. the slice owner the partitioner is asked about. */ + private val workerId: ActorVirtualIdentity = ActorVirtualIdentity("Worker:WF1-reader-main-0") + + /** A second worker on the same link, so "routed to me" is distinguishable from "routed to anyone". */ + private val siblingWorkerId: ActorVirtualIdentity = + ActorVirtualIdentity("Worker:WF1-reader-main-1") + + private val schema: Schema = + Schema().add("id", AttributeType.INTEGER).add("name", AttributeType.STRING) + + private def tuple(id: Int): Tuple = + Tuple.builder(schema).addSequentially(Array(Int.box(id), s"row-$id")).build() + + /** + * Workflow / execution ids that no other suite uses, so the tables these tests create cannot + * collide with another suite's in the shared catalog. + */ + private val workflowId = WorkflowIdentity(97311L) + private val executionId = ExecutionIdentity(97312L) + + private def baseURI(operatorId: String): URI = + VFSURIFactory.createPortBaseURI( + workflowId, + executionId, + GlobalPortIdentity( + PhysicalOpIdentity(OperatorIdentity(s"ipmrt-$operatorId"), "main"), + PortIdentity(), + input = false + ) + ) + + private def write(uri: URI, docSchema: Schema, rows: Seq[Tuple]): Unit = { + val writer = DocumentFactory + .createDocument(uri, docSchema) + .writer(s"ipmrt-${uri.getPath}") + .asInstanceOf[BufferedItemWriter[Tuple]] + writer.open() + rows.foreach(writer.putOne) + writer.close() + } + + /** + * Materializes one port: the state table and the result table the reader opens in that order. + * Both must exist — `RegionExecutionManager` creates the pair together — so even a + * states-only or tuples-only case writes both tables. + * + * @param states (user state, loop counter, LoopStart id) triples, written as state rows. + */ + private def materialize( + operatorId: String, + tuples: Seq[Tuple] = Seq.empty, + states: Seq[(State, Long, String)] = Seq.empty + ): URI = { + val base = baseURI(operatorId) + write( + VFSURIFactory.stateURI(base), + State.schema, + states.map { case (state, counter, startId) => state.toTuple(counter, startId) } + ) + write(VFSURIFactory.resultURI(base), schema, tuples) + base + } + + /** Broadcast keeps every tuple, so it isolates the replay from the partitioner filter. */ + private def broadcastTo(receivers: ActorVirtualIdentity*): Partitioning = + BroadcastPartitioning( + batchSize = 1, + channels = receivers.map(ChannelIdentity(workerId, _, isControl = false)) + ) + + private def hashShuffleTo(receivers: ActorVirtualIdentity*): Partitioning = + HashBasedShufflePartitioning( + batchSize = 1, + channels = receivers.map(ChannelIdentity(workerId, _, isControl = false)), + 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. + */ + private def replay( + uri: URI, + partitioning: Partitioning, + readerFor: ActorVirtualIdentity = workerId + ): (InputPortMaterializationReaderThread, Seq[WorkflowFIFOMessage]) = { + val queue = new LinkedBlockingQueue[DPInputQueueElement]() + val reader = + new InputPortMaterializationReaderThread(uri, queue, readerFor, partitioning) + reader.run() + (reader, drain(queue)) + } + + private def drain(queue: LinkedBlockingQueue[DPInputQueueElement]): Seq[WorkflowFIFOMessage] = { + val drained = ArrayBuffer[WorkflowFIFOMessage]() + var next = queue.poll() + while (next != null) { + drained += next.asInstanceOf[FIFOMessageElement].msg + next = queue.poll() + } + drained.toSeq + } + + private def ecmIds(messages: Seq[WorkflowFIFOMessage]): Seq[EmbeddedControlMessageIdentity] = + messages.collect { case WorkflowFIFOMessage(_, _, ecm: EmbeddedControlMessage) => ecm.id } + + private def dataFrames(messages: Seq[WorkflowFIFOMessage]): Seq[DataFrame] = + messages.collect { case WorkflowFIFOMessage(_, _, frame: DataFrame) => frame } + + private def stateFrames(messages: Seq[WorkflowFIFOMessage]): Seq[StateFrame] = + messages.collect { case WorkflowFIFOMessage(_, _, frame: StateFrame) => frame } + + private def deliveredIds(messages: Seq[WorkflowFIFOMessage]): Seq[Int] = + dataFrames(messages).flatMap(_.frame.toSeq).map(_.getField[Int]("id")) + + behavior of "InputPortMaterializationReaderThread" + + it should "bracket the replay with START_CHANNEL and END_CHANNEL and mark itself finished" in { + val uri = materialize("bracket", tuples = Seq(tuple(1))) + val (reader, 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. + assert( + ecmIds(messages) == Seq( + EmbeddedControlMessageIdentity("StartChannel"), + EmbeddedControlMessageIdentity("EndChannel") + ) + ) + assert(messages.head.payload.isInstanceOf[EmbeddedControlMessage]) + assert(messages.last.payload.isInstanceOf[EmbeddedControlMessage]) + // START is unaligned (it only opens the channel) while END must align the port, which is what + // lets the worker decide the port is complete. + val Seq(start, end) = messages.collect { + case WorkflowFIFOMessage(_, _, ecm: EmbeddedControlMessage) => ecm + } + assert(start.ecmType == NO_ALIGNMENT) + assert(end.ecmType == PORT_ALIGNMENT) + assert(reader.finished) + } + + it should "not report itself finished before the replay runs" in { + val queue = new LinkedBlockingQueue[DPInputQueueElement]() + val reader = 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)) + + // 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). + val expectedChannel = ChannelIdentity( + getFromActorIdForInputPortStorage(uri.toString, workerId), + workerId, + isControl = false + ) + assert(messages.map(_.channelId).distinct == Seq(expectedChannel)) + // FIFO ordering downstream is by sequence number, so they must be dense and start at 0. + assert(messages.map(_.sequenceNumber) == messages.indices.map(_.toLong)) + } + + it should "replay states ahead of tuples, carrying each row's loop envelope" in { + // Distinct counters and ids per state, so a reader that reused one row's envelope for the + // other, or defaulted to the "no loop" values, would show up. + val states = Seq( + (State(Map("phase" -> "a")), 7L, "loop-start-A"), + (State(Map("phase" -> "b")), 9L, "loop-start-B") + ) + val uri = materialize("states", tuples = Seq(tuple(1), tuple(2)), states = states) + + 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. + val lastState = messages.lastIndexWhere(_.payload.isInstanceOf[StateFrame]) + val firstData = messages.indexWhere(_.payload.isInstanceOf[DataFrame]) + assert(lastState >= 0 && firstData >= 0 && lastState < firstData) + assert( + stateFrames(messages).map(frame => + (frame.frame.values("phase"), frame.loopCounter, frame.loopStartId) + ) == Seq(("a", 7L, "loop-start-A"), ("b", 9L, "loop-start-B")) + ) + } + + 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)) + + assert(stateFrames(messages).isEmpty) + assert(deliveredIds(messages) == Seq(1)) + } + + it should "deliver every tuple broadcast to it, in materialization order, in one batch" in { + val ids = 1 to 5 + val uri = materialize("broadcast", tuples = ids.map(tuple)) + + val (_, messages) = replay(uri, broadcastTo(workerId, siblingWorkerId)) + + // Under the batch size, everything leaves in the single trailing flush. + assert(dataFrames(messages).size == 1) + assert(deliveredIds(messages) == ids) + } + + it should "drop every tuple the partitioning routes to another worker, but still bracket the channel" in { + val uri = materialize("elsewhere", tuples = (1 to 5).map(tuple)) + + // 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)) + + assert(deliveredIds(messages).isEmpty) + assert(dataFrames(messages).isEmpty) + // An empty slice is still a complete channel — the port would never finish otherwise. + assert( + ecmIds(messages) == Seq( + EmbeddedControlMessageIdentity("StartChannel"), + EmbeddedControlMessageIdentity("EndChannel") + ) + ) + assert(reader.finished) + } + + it should "split the materialization between the workers of a hash-shuffled link" in { + val ids = 1 to 20 + 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) + + // 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 + // the `exists` would deliver none. + assert(deliveredIds(mine).nonEmpty) + assert(deliveredIds(siblings).nonEmpty) + assert(deliveredIds(mine).intersect(deliveredIds(siblings)).isEmpty) + assert((deliveredIds(mine) ++ deliveredIds(siblings)).sorted == ids) + } + + it should "cut a full batch as soon as the buffer reaches the transfer batch size" in { + // One tuple past a full batch, so the split is observable and the remainder is not itself a + // full batch: a reader that only flushed at the end would produce a single oversized frame. + val batchSize = ApplicationConfig.defaultDataTransferBatchSize + val ids = 1 to (batchSize + 1) + val uri = materialize("batching", tuples = ids.map(tuple)) + + val (_, messages) = replay(uri, broadcastTo(workerId)) + + assert(dataFrames(messages).map(_.frame.length) == Seq(batchSize, 1)) + assert(deliveredIds(messages) == ids) + } + + it should "wrap a storage failure in a RuntimeException that keeps the cause" in { + // Nothing was ever materialized at this URI, so opening the state document fails. The reader + // runs on its own thread with no supervisor, so it re-throws rather than swallowing. + val uri = baseURI("missing") + val queue = new LinkedBlockingQueue[DPInputQueueElement]() + val reader = + new InputPortMaterializationReaderThread(uri, queue, workerId, broadcastTo(workerId)) + + val failure = intercept[RuntimeException] { + reader.run() + } + + 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... + assert(ecmIds(drain(queue)) == Seq(EmbeddedControlMessageIdentity("StartChannel"))) + // ...but the channel never completes, which is what stops the port being called complete. + assert(!reader.finished) + } +}
