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-7269-f3231c8d1b7b664ce6a3f716423b892f6cd329aa in repository https://gitbox.apache.org/repos/asf/texera.git
commit 4f66fd580bfbf6d0ea56902c6ee04103bf06b2a8 Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Aug 3 12:38:10 2026 -0700 test(amber): cover WorkflowActor.setupReplay and the Coordinator actor (#7269) ### What changes were proposed in this PR? Two engine classes where the existing specs stopped short: `WorkflowActor` (26 missed, 44.9%) and `Coordinator` (26 missed, 66.0%, no spec at all). **`WorkflowActorSpec` gains four cases, all around `setupReplay`.** Both branches are drivable with a `ram:///` log: *Replay-from-scratch.* Seed `ProcessingStep(dataCid, -1)`, `ProcessingStep(controlCid, 0)` and two `MessageContent` records, then call `setupReplay` directly. The discriminating assertion: ``` NetworkInputGateway.tryPickChannel ──prefers──▶ CONTROL but returns ──▶ DATA ``` Data can only come back first if `setupReplay` installed a `ReplayOrderEnforcer` seeded from `logManager.getStep`, which pins step −1 to the data channel. Drop the `addEnforcer` call, or start the enforcer at 0 instead of the log manager's step, and the control channel comes back instead. The replayed payload carries a marker string that exists nowhere but the log this test wrote, so provenance is established rather than type-checked. *Checkpoint branch.* Selected by an empty log file inside the destination subfolder. It asserts **branch selection only** — that `loadFromCheckpoint` fired and the top-level log's messages were *not* injected (both are seeded, so the branches are distinguishable). It deliberately does not assert the loaded value: `CheckpointState` does not extend `java.io.Serializable` while `cluster.conf` sets `allow-java-serialization = off`, so a real one can be neither written nor read today. Pinning the resulting null would cement that bug and break whoever fixes the binding. Plus the cluster-address closure and the `receiveMessageAndAck` catch, whose contract is to log **and rethrow** so the supervisor sees the failure. **`CoordinatorSpec` is new.** A real `Coordinator` is constructible in a unit spec — `WorkflowSchedulerSpec` already proves `updateSchedule` runs with no DB, and `DefaultCostEstimator` swallows an uninitialised `SqlServer` via `Try`. Covers the replay-status callbacks, `initState`'s restore branch, `loadFromCheckpoint`, and the supervisor strategy. The `loadFromCheckpoint` case is the interesting one: it asserts both that `cp` is swapped for a different instance **and** that an `ExecutionStatsUpdate` still reaches the parent. The second half is what pins the `cp.outputHandler = logManager.sendCommitted` re-attach — the handler is `@transient`, so without the re-attach it is null after the kryo round-trip and the send NPEs. Left alone on purpose: `CoordinatorConfig.default` and `Coordinator.props`, both already executed by the untagged e2e specs, where any test would reduce to asserting `ApplicationConfig` constants against themselves; and `loadFromCheckpoint`'s worker-revival loop, which needs real workers and DP threads. A trap worth recording for the next person: `SequentialRecordReader.mkRecordIterator` uses a `lazy val input`, and its catch handler re-forces the failed lazy val, so a **missing** file throws out of the iterator while an **empty** one yields null. A spec has to create the file, not merely the folder, or it gets a green-looking test that passed for the wrong reason. Each `ram://` folder is unique per case, since `VFS.getManager` is a JVM-wide singleton. No production file is touched. ### Any related issues, documentation, discussions? Closes #7267 ### How was this PR tested? 9 new tests (4 appended to `WorkflowActorSpec`, 5 in the new `CoordinatorSpec`) — 19 tests across the two suites, Java 17: ``` sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.common.WorkflowActorSpec org.apache.texera.amber.engine.architecture.coordinator.CoordinatorSpec" ``` ``` [info] Suites: completed 2, aborted 0 [info] Tests: succeeded 19, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. ``` Also run in one JVM alongside the neighbouring replay/checkpoint suites (`WorkflowSchedulerSpec`, `GlobalReplayManagerSpec`, `TakeGlobalCheckpointHandlerSpec`, `ReplayLogGeneratorSpec`, `LoggingSpec`, `WorkflowWorkerSpec`, `CheckpointSubsystemSpec`) to confirm the `ram://` folders and the shared `SessionState` registry do not interfere. `Test/scalafmtCheck` and `Test/scalafix --check` both `[success]`. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../architecture/common/WorkflowActorSpec.scala | 255 ++++++++++++++++++++- .../architecture/coordinator/CoordinatorSpec.scala | 236 ++++++++++++++++++- 2 files changed, 488 insertions(+), 3 deletions(-) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActorSpec.scala index 447245be77..c4a6696e95 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActorSpec.scala @@ -21,7 +21,12 @@ package org.apache.texera.amber.engine.architecture.common import org.apache.pekko.actor.{ActorSystem, Props, UnhandledMessage} import org.apache.pekko.testkit.{TestActorRef, TestKit, TestProbe} -import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ChannelIdentity, + EmbeddedControlMessageIdentity +} import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{ CreditRequest, CreditResponse, @@ -31,14 +36,30 @@ import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{ RegisterActorRef } import org.apache.texera.amber.engine.architecture.control.utils.TrivialControlTester +import org.apache.texera.amber.engine.architecture.logreplay.{ + MessageContent, + ProcessingStep, + ReplayLogRecord +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ControlInvocation, + EmptyRequest +} import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ MainThreadDelegateMessage, + StateRestoreConfig, TriggerSend } +import org.apache.texera.amber.engine.common.CheckpointState +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} import org.apache.texera.amber.engine.common.ambermessage.{DataFrame, WorkflowFIFOMessage} +import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpecLike +import java.net.URI +import java.util.concurrent.atomic.AtomicInteger import scala.concurrent.duration.DurationInt class WorkflowActorSpec @@ -46,11 +67,25 @@ class WorkflowActorSpec with AnyFlatSpecLike with BeforeAndAfterAll { + override def beforeAll(): Unit = { + // WorkflowActor's getAvailableNodeAddressesFunc asks "/user/cluster-info"; + // without a listener there that ask times out. + system.actorOf(Props[SingleNodeListener](), "cluster-info") + } + override def afterAll(): Unit = { TestKit.shutdownActorSystem(system) } private val selfId: ActorVirtualIdentity = ActorVirtualIdentity("self-worker") + + // Carries a marker only this spec writes, so a replayed data payload can be traced back to the + // log rather than merely type-checked. + private val replayedTuple: Tuple = + Tuple + .builder(Schema().add("marker", AttributeType.STRING)) + .add("marker", AttributeType.STRING, "replayed-data") + .build() private val otherId: ActorVirtualIdentity = ActorVirtualIdentity("other-worker") // A control channel whose destination (toWorkerId) is the tester actor itself. @@ -89,6 +124,45 @@ class WorkflowActorSpec name ) + private val replayDestination: EmbeddedControlMessageIdentity = + EmbeddedControlMessageIdentity("workflow-actor-spec-replay-to") + + /** Write `records` as the replay log named `logFileName` under `folder`. + * + * Each case uses its own `ram://` folder because the VFS manager is a + * JVM-wide singleton. + */ + private def writeLog(folder: URI, logFileName: String, records: Seq[ReplayLogRecord]): Unit = { + val writer = + SequentialRecordStorage.getStorage[ReplayLogRecord](Some(folder)).getWriter(logFileName) + try { + records.foreach(writer.writeRecord) + writer.flush() + } finally { + writer.close() + } + } + + /** Counts `loadFromCheckpoint` calls so the checkpoint branch of `setupReplay` + * is observable without depending on what the (currently unreadable) + * checkpoint record deserializes to. + */ + private class CheckpointCountingTester(workerId: ActorVirtualIdentity) + extends TrivialControlTester(workerId) { + val loadFromCheckpointCalls = new AtomicInteger() + + override def loadFromCheckpoint(chkpt: CheckpointState): Unit = { + loadFromCheckpointCalls.incrementAndGet() + } + } + + /** Fails every input message, to exercise `receiveMessageAndAck`'s catch. */ + private class FailingInputTester(workerId: ActorVirtualIdentity) + extends TrivialControlTester(workerId) { + override def handleInputMessage(messageId: Long, workflowMsg: WorkflowFIFOMessage): Unit = + throw new IllegalStateException("handleInputMessage failed") + } + // --------------------------------------------------------------------------- // receiveCreditMessages (WorkflowActor lines 151-155) // --------------------------------------------------------------------------- @@ -251,4 +325,183 @@ class WorkflowActorSpec val forwarded = parent.expectMsgType[GetActorRef] assert(forwarded.id == unknownDest) } + + // --------------------------------------------------------------------------- + // getAvailableNodeAddressesFunc (WorkflowActor lines 89-97) + // --------------------------------------------------------------------------- + + it should "resolve cluster node addresses by asking /user/cluster-info" in { + val parent = TestProbe() + val ref = newTester(parent, "cluster-addresses") + + // PekkoActorService ships with `() => Array.empty`; the WorkflowActor + // constructor replaces it with an ask against "/user/cluster-info". + // ExecutorDeployment.createWorkers round-robins workers over whatever this + // returns and asserts it is non-empty ("no available computation nodes"), + // so losing the assignment breaks every deployment. + val addresses = ref.underlyingActor.actorService.getClusterNodeAddresses + + // SingleNodeListener answers with its own path address, which for a local + // (non-remote) ActorSystem is the address shared by every local actor. + assert(addresses.toSeq == Seq(parent.ref.path.address)) + } + + // --------------------------------------------------------------------------- + // receiveMessageAndAck failure path (WorkflowActor lines 138-146) + // --------------------------------------------------------------------------- + + it should "register the sender and then rethrow when handleInputMessage fails" in { + val parent = TestProbe() + val messageSender = TestProbe() + val ref = TestActorRef[FailingInputTester]( + Props(new FailingInputTester(selfId)), + parent.ref, + "input-message-throws" + ) + val upstreamId = ActorVirtualIdentity("upstream") + assert(!ref.underlyingActor.actorRefMappingService.hasActorRef(upstreamId)) + + // `TestActorRef.receive` runs the behavior directly instead of going through + // the mailbox, so supervision does not swallow the failure and `intercept` + // sees whatever escapes `receiveMessageAndAck`. The contract under test is + // that the handler's exception is re-thrown rather than logged-and-eaten: + // the coordinator's AllForOneStrategy can only report a FatalError if the + // failure actually reaches it. + val thrown = intercept[IllegalStateException] { + ref.receive(networkMessageTo(selfId), messageSender.ref) + } + assert(thrown.getMessage == "handleInputMessage failed") + + // The mapping is updated before the handler runs, so a failed message still + // leaves a route back to its sender. + assert(ref.underlyingActor.actorRefMappingService.getActorRef(upstreamId) == messageSender.ref) + } + + // --------------------------------------------------------------------------- + // setupReplay (WorkflowActor lines 190-227) + // --------------------------------------------------------------------------- + + it should "replay from scratch by injecting logged messages behind the logged channel order" in { + val parent = TestProbe() + val ref = newTester(parent, "setup-replay-scratch") + val actor = ref.underlyingActor + val gateway = actor.ap.inputGateway + + val dataCid = channelTo(selfId, isControl = false) + val controlCid = channelTo(selfId, isControl = true) + val readFrom = new URI("ram:///workflow-actor-spec-replay-from-scratch/") + writeLog( + readFrom, + actor.getLogName, + Seq( + // ProcessingStepCursor.INIT_STEP is -1 and the log manager has not + // stepped yet, so the first record has to carry step -1; with any + // larger value ReplayOrderEnforcer never latches a current channel and + // nothing is ever pickable. + ProcessingStep(dataCid, -1L), + ProcessingStep(controlCid, 0L), + MessageContent(WorkflowFIFOMessage(dataCid, 0L, DataFrame(Array(replayedTuple)))), + MessageContent( + WorkflowFIFOMessage( + controlCid, + 0L, + ControlInvocation( + "replayed-control", + EmptyRequest(), + AsyncRPCContext(otherId, selfId), + 0 + ) + ) + ) + ) + ) + + val completions = new AtomicInteger() + actor.setupReplay( + actor.ap, + StateRestoreConfig(readFrom, replayDestination), + () => completions.incrementAndGet() + ) + + // NetworkInputGateway.tryPickChannel prefers CONTROL channels, and both + // channels hold a replayed message. The DATA channel can only come back + // first because setupReplay installed a ReplayOrderEnforcer seeded from + // logManager.getStep, which pins step -1 to the data channel. Dropping the + // addEnforcer call, or starting the enforcer at step 0 instead of the log + // manager's step, both hand back the control channel here. + val first = gateway.tryPickChannel + assert(first.map(_.channelId).contains(dataCid)) + assert(completions.get() == 0, "replay must not be complete while a step is still pending") + + val replayedData = first.get.take + // The marker string exists nowhere but the log this test wrote, so this establishes + // provenance -- a bare `isInstanceOf[DataFrame]` would not, since any DataFrame satisfies it. + assert(replayedData.payload match { + case DataFrame(frame) => frame.toSeq.map(_.getField[String]("marker")) == Seq("replayed-data") + case other => fail(s"unexpected replayed payload: $other") + }) + + // Processing that message advances the cursor to step 0, which is what + // releases the next logged step (the control channel). + actor.logManager.withFaultTolerant(dataCid, None) { + // the replayed data message is the "work" for this step + } + + val second = gateway.tryPickChannel + assert(second.map(_.channelId).contains(controlCid)) + // The enforcer's queue is now drained, so replay is over -- exactly once. + assert(completions.get() == 1) + assert(second.get.take.payload match { + case invocation: ControlInvocation => invocation.methodName == "replayed-control" + case other => fail(s"unexpected replayed payload: $other") + }) + } + + it should "take the checkpoint branch of setupReplay when the destination subfolder exists" in { + val parent = TestProbe() + val ref = TestActorRef[CheckpointCountingTester]( + Props(new CheckpointCountingTester(selfId)), + parent.ref, + "setup-replay-checkpoint" + ) + val actor = ref.underlyingActor + + val readFrom = new URI("ram:///workflow-actor-spec-replay-checkpoint/") + // Seed the top-level log as well, so the two branches are distinguishable: + // had setupReplay taken the from-scratch branch it would have injected this + // message into the input gateway. + writeLog( + readFrom, + actor.getLogName, + Seq( + ProcessingStep(channelTo(selfId, isControl = false), -1L), + MessageContent( + WorkflowFIFOMessage(channelTo(selfId, isControl = false), 0L, DataFrame(Array.empty)) + ) + ) + ) + // Creating a checkpoint storage at the per-destination subfolder creates + // that folder, which is what `containsFolder` keys on. The record file has + // to exist too (empty is fine): SequentialRecordReader opens it through a + // `lazy val`, and on a missing file the catch handler re-forces the failed + // lazy val and throws straight out of the iterator. + SequentialRecordStorage + .getStorage[CheckpointState](Some(readFrom.resolve(replayDestination.toString))) + .getWriter(actor.getLogName) + .close() + + actor.setupReplay(actor.ap, StateRestoreConfig(readFrom, replayDestination), () => ()) + + // Only branch selection is asserted. The record read back from an empty + // checkpoint file is null, and CheckpointState is not java.io.Serializable + // while cluster.conf turns java serialization off, so no real + // CheckpointState can be written or read today; pinning the null would + // cement that. + assert(actor.loadFromCheckpointCalls.get() == 1) + assert( + actor.ap.inputGateway.getAllChannels.isEmpty, + "the checkpoint branch must not replay the top-level log" + ) + } + } diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/CoordinatorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/CoordinatorSpec.scala index a9328563e1..6392a90cbb 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/CoordinatorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/CoordinatorSpec.scala @@ -19,16 +19,60 @@ package org.apache.texera.amber.engine.architecture.coordinator -import org.apache.pekko.actor.{ActorSystem, Props} -import org.apache.pekko.testkit.{ImplicitSender, TestKit} +import org.apache.pekko.actor.{Actor, ActorSystem, Props} +import org.apache.pekko.testkit.{ImplicitSender, TestActorRef, TestKit, TestProbe} import org.apache.pekko.util.Timeout import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ChannelIdentity, + EmbeddedControlMessageIdentity +} +import org.apache.texera.amber.core.workflow.{PhysicalPlan, PortIdentity, WorkflowContext} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{ + NetworkMessage, + RegisterActorRef +} +import org.apache.texera.amber.engine.architecture.logreplay.ReplayLogRecord +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.StateRestoreConfig +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage +import org.apache.texera.amber.engine.common.virtualidentity.util.{CLIENT, COORDINATOR} +import org.apache.texera.amber.engine.common.{CheckpointState, SerializedState} +import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpecLike +import java.net.URI import scala.concurrent.ExecutionContextExecutor import scala.concurrent.duration._ +/** + * Unit coverage for the `Coordinator` actor's recovery, checkpoint-restore and + * supervision plumbing -- the parts the untagged e2e specs never reach because + * they only ever run a workflow start-to-finish. + * + * The breakages this spec catches: + * - losing either `GlobalReplayManager` callback, which is how the client + * learns that a workflow entered / left recovery. + * - dropping `initState`'s state-restore branch, so a coordinator asked to + * recover silently starts clean instead. + * - `loadFromCheckpoint` forgetting to install the checkpointed processor, to + * re-attach its `@transient` runtime services, or to re-attach + * `cp.outputHandler` (which is `@transient` too, so the first send after a + * restore would NPE). + * - weakening the `AllForOneStrategy`, which is the only place a child's + * fatal error is reported to the client. + * + * A real `Coordinator` is constructible here: `WorkflowScheduler.updateSchedule` + * needs no database (`DefaultCostEstimator` swallows an uninitialised + * `SqlServer`), and `SessionState.getAllSessionStates` is empty in a unit JVM so + * the region fan-out is a no-op. Replay/checkpoint storage is the in-memory + * `ram://` VFS also used by `LoggingSpec`, with a distinct folder per case + * because the VFS manager is a JVM-wide singleton. + */ class CoordinatorSpec extends TestKit(ActorSystem("CoordinatorSpec")) with ImplicitSender @@ -46,6 +90,194 @@ class CoordinatorSpec TestKit.shutdownActorSystem(system) } + private val workerA = ActorVirtualIdentity("Worker:WF1-E1-op-layer-0") + private val workerB = ActorVirtualIdentity("Worker:WF1-E1-op-layer-1") + private val replayDestination = EmbeddedControlMessageIdentity("coordinator-spec-replay-to") + + /** A two-operator CSV plan -- the same fixture `WorkflowSchedulerSpec` uses. */ + private def newPhysicalPlan(): (WorkflowContext, PhysicalPlan) = { + val csvOp = TestOperators.headerlessSmallCsvScanOpDesc() + val keywordOp = TestOperators.keywordSearchOpDesc("column-1", "Asia") + val workflow = buildWorkflow( + List(csvOp, keywordOp), + List( + LogicalLink( + csvOp.operatorIdentifier, + PortIdentity(0), + keywordOp.operatorIdentifier, + PortIdentity(0) + ) + ), + new WorkflowContext() + ) + (workflow.context, workflow.physicalPlan) + } + + private def newCoordinator( + parent: TestProbe, + config: CoordinatorConfig, + name: String + ): TestActorRef[Coordinator] = { + val (context, plan) = newPhysicalPlan() + TestActorRef[Coordinator](Coordinator.props(context, plan, config), parent.ref, name) + } + + /** Throws on any message, to drive the coordinator's supervisor strategy. */ + private class ExplodingChild extends Actor { + override def receive: Receive = { + case _ => throw new IllegalStateException("child exploded") + } + } + + // --------------------------------------------------------------------------- + // handleReplayMessages + the two GlobalReplayManager callbacks (lines 105-113, + // 187-190) + // --------------------------------------------------------------------------- + + "Coordinator" should "report recovery start and completion to the client on ReplayStatusUpdate" in { + val parent = TestProbe() + val coordinator = newCoordinator(parent, CoordinatorConfig.default, "coordinator-replay-status") + parent.expectMsgType[RegisterActorRef] + + // First worker entering recovery -> the onStart callback fires. + coordinator ! ReplayStatusUpdate(workerA, status = true) + parent.expectMsg(WorkflowRecoveryStatus(true)) + + // A second worker joining an already-recovering workflow must not + // re-announce the start, and the first one clearing must not announce the + // end while the second is still recovering. + coordinator ! ReplayStatusUpdate(workerB, status = true) + coordinator ! ReplayStatusUpdate(workerA, status = false) + parent.expectNoMessage(200.millis) + + // Last worker out -> the onComplete callback fires. + coordinator ! ReplayStatusUpdate(workerB, status = false) + parent.expectMsg(WorkflowRecoveryStatus(false)) + } + + // --------------------------------------------------------------------------- + // initState's state-restore branch (lines 130-141) + // --------------------------------------------------------------------------- + + it should "announce recovery around the replay it sets up at startup" in { + val parent = TestProbe() + val readFrom = new URI("ram:///coordinator-spec-init-state-restore/") + // An EMPTY log named after the coordinator. The file itself has to exist: + // SequentialRecordReader opens it through a `lazy val`, and on a missing + // file the catch handler re-forces the failed lazy val and throws straight + // out of the iterator. + SequentialRecordStorage + .getStorage[ReplayLogRecord](Some(readFrom)) + .getWriter(COORDINATOR.name) + .close() + + newCoordinator( + parent, + CoordinatorConfig.default.copy( + stateRestoreConfOpt = Some(StateRestoreConfig(readFrom, replayDestination)) + ), + "coordinator-init-state-restore" + ) + + // Both callbacks fire inside initState, i.e. before preStart registers the + // coordinator's own actor ref -- the ordering pins that this ran during + // startup rather than being triggered by anything the test sent. + parent.expectMsg(WorkflowRecoveryStatus(true)) + parent.expectMsg(WorkflowRecoveryStatus(false)) + parent.expectMsgType[RegisterActorRef] + } + + it should "not announce recovery when no state-restore config is given" in { + val parent = TestProbe() + newCoordinator(parent, CoordinatorConfig.default, "coordinator-init-state-plain") + + // Only the actor-ref registration; the restore branch stayed shut. + parent.expectMsgType[RegisterActorRef] + parent.expectNoMessage(200.millis) + } + + // --------------------------------------------------------------------------- + // loadFromCheckpoint (lines 223-251) + // --------------------------------------------------------------------------- + + it should "install the checkpointed processor, re-attach its services and report stats" in { + val parent = TestProbe() + val coordinator = + newCoordinator(parent, CoordinatorConfig.default, "coordinator-load-checkpoint") + val actor = coordinator.underlyingActor + val liveProcessor = actor.cp + parent.expectMsgType[RegisterActorRef] + + // Give the checkpointed processor output-FIFO state the live one does not + // have, so "the restored cp came from the checkpoint" is distinguishable + // from "the restored cp is some fresh processor". + val markerChannel = + ChannelIdentity(COORDINATOR, ActorVirtualIdentity("checkpointed-peer"), isControl = true) + val checkpointed = + new CoordinatorProcessor( + new WorkflowContext(), + CoordinatorConfig.default, + COORDINATOR, + _ => () + ) + (0 until 3).foreach(_ => checkpointed.outputGateway.getSequenceNumber(markerChannel)) + assert(!liveProcessor.outputGateway.getFIFOState.contains(markerChannel)) + + val chkpt = new CheckpointState() + chkpt.save(SerializedState.CP_STATE_KEY, checkpointed) + // No un-acked output to resend, and a fresh WorkflowExecution has no running + // region executions, so the worker-revival loop is a no-op here. + chkpt.save(SerializedState.OUTPUT_MSG_KEY, Array.empty[WorkflowFIFOMessage]) + + actor.loadFromCheckpoint(chkpt) + + assert(actor.cp ne liveProcessor) + assert(actor.cp.outputGateway.getFIFOState.get(markerChannel).contains(3L)) + + // Every service handle on CoordinatorProcessor is `@transient`, so Kryo + // hands them back null; attachRuntimeServicesToCPState re-wires them to the + // actor's live instances. + assert(actor.cp.logManager eq actor.logManager) + assert(actor.cp.actorService eq actor.actorService) + assert(actor.cp.transferService eq actor.transferService) + assert(actor.cp.actorRefService eq actor.actorRefMappingService) + + // `outputHandler` is `@transient` too. Without the + // `cp.outputHandler = logManager.sendCommitted` re-attach, this very send + // NPEs on a null handler, so seeing the stats event arrive at the client + // pins that line. + parent.fishForMessage(5.seconds) { + case NetworkMessage(_, WorkflowFIFOMessage(channel, _, _: ExecutionStatsUpdate)) => + channel.toWorkerId == CLIENT + case _ => false + } + } + + // --------------------------------------------------------------------------- + // supervisorStrategy (lines 204-213) + // --------------------------------------------------------------------------- + + it should "stop a failing child and report the failure to the client as a FatalError" in { + val parent = TestProbe() + val coordinator = newCoordinator(parent, CoordinatorConfig.default, "coordinator-supervisor") + parent.expectMsgType[RegisterActorRef] + + val child = coordinator.underlyingActor.context.actorOf(Props(new ExplodingChild), "boom") + watch(child) + child ! "explode" + + // maxNrOfRetries = 0 with Stop: the child is not restarted, it is killed. + expectTerminated(child, 5.seconds) + + // The child runs on the default dispatcher, so the FatalError arrives + // asynchronously relative to this thread. + parent.fishForMessage(5.seconds) { + case NetworkMessage(_, WorkflowFIFOMessage(_, _, FatalError(e, _))) => + e.getMessage == "child exploded" + case _ => false + } + } + // private val logicalPlan1 = // """{ // |"operators":[
