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/03 19:58:55 UTC
(texera) branch gh-readonly-queue/main/pr-7269-f3231c8d1b7b664ce6a3f716423b892f6cd329aa created (now 4f66fd580b)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7269-f3231c8d1b7b664ce6a3f716423b892f6cd329aa
in repository https://gitbox.apache.org/repos/asf/texera.git
at 4f66fd580b test(amber): cover WorkflowActor.setupReplay and the Coordinator actor (#7269)
This branch includes the following new commits:
new 4f66fd580b test(amber): cover WorkflowActor.setupReplay and the Coordinator actor (#7269)
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: test(amber): cover WorkflowActor.setupReplay and the Coordinator actor (#7269)
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/main/pr-7269-f3231c8d1b7b664ce6a3f716423b892f6cd329aa
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 4f66fd580bfbf6d0ea56902c6ee04103bf06b2a8
Author: Xinyuan Lin <xi...@uci.edu>
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":[