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-7770-80913ed92a4be5d198eb3d5eeaf80bcdc198bad3 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2d60462a9527e5d551b53a94b7875cacfc8e5018 Author: Xinyuan Lin <[email protected]> AuthorDate: Wed Aug 19 06:14:26 2026 +0000 test(amber): cover the promise handlers, metadata endpoint and actor-ref mapping (#7770) ### What changes were proposed in this PR? Six small amber files, three of which had no spec at all. 12 new tests across 4 new spec files and 2 extended ones. | File | Before | After | |---|---|---| | `DebugCommandHandler` | 33.3% (1/3) | **100%** (3/3) | | `ConsoleMessageHandler` | 33.3% (1/3) | **100%** (3/3) | | `SystemMetadataResource` | **0%** (0/2) | **100%** (2/2) | | `StartHandler` | 73.9%, 2 missed + 4 partial | **87.0%**, 0 missed | | `PrepareCheckpointHandler` | 61.9%, 2 missed + 6 partial | **71.4%**, 0 missed | | `PekkoActorRefMappingService` | 85.7%, 2 missed + 5 partial | **87.8%**, 0 missed | **This is 12 lines, and I would rather say so than dress it up.** What earns it is that every one is live code, three files go from partly- or wholly-untested to 100%, none of it needs infrastructure, and all 10 mutations die. The `PrepareCheckpointHandler` case also closes the prepare-to-finalize handshake gap that file's own scaladoc flagged as a follow-up — it is now driven against a real `WorkflowWorker` via `TestKit` rather than stopping at the main-thread hand-off. ### Verification 10 mutations, **10 killed, no survivors**. Each was applied alone by a driver that asserts the anchor occurs exactly once in both directions, reverted with `git diff -- '*/src/main/*'` confirmed empty afterwards, and the failing test read by name from `amber/target/test-reports`. | Mutation | Killed by | |---|---| | `DebugCommandHandler` addresses the sender instead of the named worker | addresses the command to the worker the request names, not to the sender | | **exchange** the debug request's two fields | relays the request body verbatim | | **exchange** `source` and `title` on the console message | hands the client the reported message unchanged | | **exchange** the two ends of the client channel in `sendToClient` | puts the message on the client channel and nowhere else | | `SystemMetadataResource` filters out `Filter` operators | hands back an entry for every operator the workspace can place | | `SystemMetadataResource` reverses the group order | hands back the palette's group order | | `StartHandler`'s reader-thread check `nonEmpty` -> `isEmpty` | two tests, incl. refuses a worker that is neither a source nor a reader of materialized input | | `PekkoActorRefMappingService` registers the id before the parent send | swallows a failed parent lookup and still asks again for the same id | | **exchange** the checkpoint's queued-input and output keys | saves the worker's queued input and starts recording what arrives next | | drop the recorded-inputs write | same test | ### Four sibling files were assessed and deliberately left alone Recording the evidence so this is not re-derived: - **`PekkoActorService`** — its two uncovered lines are the entire bodies of `sendToSelfOnce` and `ask`, and both have **zero call sites** anywhere in main, test, bench or scripts. Dead code. (Its other originally-missed lines are in fact already covered.) - **`WorkflowMessage`** — `case _ => 200L` is unreachable: the trait is `sealed` and has a single subtype in the file, so only a `null` argument could reach the default arm. Its other line is a case-class declaration whose 28 uncovered branches are scalac-generated `equals`/`copy`/`productElement`. - **`RecoveryPayload`** — the classes are already constructed by `AmberMessageEnvelopesSpec`; the uncovered remainder is purely scalac-generated members, so nothing hand-written could be deleted to make a test fail. - **`ComputingUnitWorker`** — the uncovered lines are `main`'s body, and `AmberRuntime.startActorWorker` binds a cluster seed and mutates `AmberConfig.masterNodeAddr`, which is global state in amber's shared test JVM. Within the accepted files, the remaining partials are `logger.info`/`logger.debug` macro guards (which do not execute under CI's `TEXERA_SERVICE_LOG_LEVEL=WARN`) and `val (a, b) = ...` tuple-destructuring `MatchError` arms. ### A production fragility, reported and not pinned `PekkoActorRefMappingService.retrieveActorRef`'s catch block builds its warning as `s"... parentRef = " + actorService.parent` — it re-reads the very thing whose failure it is handling. A parent lookup that fails *persistently* rather than transiently therefore throws out of the handler that exists to contain it. The new test injects exactly one failure for that reason, and says so in its scaladoc, so a fix is not blocked. No production file is touched. ### Any related issues, documentation, discussions? Closes #7769 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.coordinator.promisehandlers.DebugCommandHandlerSpec org.apache.texera.amber.engine.architecture.coordinator.promisehandlers.ConsoleMessageHandlerSpec org.apache.texera.amber.engine.architecture.worker.promisehandlers.StartHandlerSpec org.apache.texera.web.resource.SystemMetadataResourceSpec org.apache.texera.amber.engine.architecture.common.PekkoActorRefMappingServi [...] ``` ``` [info] Total number of tests run: 20 [info] Tests: succeeded 20, failed 0, canceled 0, ignored 0, pending 0 ``` The amber unit suite goes from 1919 to 1931 tests. `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) --- .../common/PekkoActorRefMappingServiceSpec.scala | 60 +++++++-- .../ConsoleMessageHandlerSpec.scala | 134 +++++++++++++++++++++ .../promisehandlers/DebugCommandHandlerSpec.scala | 132 ++++++++++++++++++++ .../PrepareCheckpointHandlerSpec.scala | 128 +++++++++++++++++--- .../worker/promisehandlers/StartHandlerSpec.scala | 128 ++++++++++++++++++++ .../web/resource/SystemMetadataResourceSpec.scala | 66 ++++++++++ 6 files changed, 623 insertions(+), 25 deletions(-) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala index 6b5cf140ac..d836008dac 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingServiceSpec.scala @@ -19,7 +19,7 @@ package org.apache.texera.amber.engine.architecture.common -import org.apache.pekko.actor.{Actor, ActorSystem, Props} +import org.apache.pekko.actor.{Actor, ActorContext, ActorRef, ActorSystem, Props} import org.apache.pekko.testkit.{TestActorRef, TestKit, TestProbe} import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{ @@ -73,18 +73,17 @@ class PekkoActorRefMappingServiceSpec WorkflowFIFOMessage(channelTo(destination), sequenceNumber, DataFrame(Array.empty)) ) + private def newContext(parent: TestProbe): ActorContext = + TestActorRef[ActorRefMappingServiceContextHolder]( + Props(new ActorRefMappingServiceContextHolder), + parent.ref, + s"actor-ref-mapping-context-${contextCounter.incrementAndGet()}" + ).underlyingActor.context + private def newActorService( id: ActorVirtualIdentity, parent: TestProbe - ): PekkoActorService = { - val contextHolder = - TestActorRef[ActorRefMappingServiceContextHolder]( - Props(new ActorRefMappingServiceContextHolder), - parent.ref, - s"actor-ref-mapping-context-${contextCounter.incrementAndGet()}" - ) - new PekkoActorService(id, contextHolder.underlyingActor.context) - } + ): PekkoActorService = new PekkoActorService(id, newContext(parent)) "askForCredit" should "forward a request only when the destination ref is known" in { val parent = TestProbe() @@ -165,9 +164,50 @@ class PekkoActorRefMappingServiceSpec assert(!service.hasActorRef(destination)) assert(service.findActorVirtualIdentity(registered.ref).isEmpty) } + + "retrieveActorRef" should "swallow a failed parent lookup and still ask again for the same id" in { + val parent = TestProbe() + val destination = ActorVirtualIdentity("unreachable-parent-destination") + val waiter = TestProbe() + // Pekko itself does not fail here -- `context.parent` reads a field and `!` never throws -- so + // the failure is injected, and only for the first read: the catch block's own log line reads + // `actorService.parent` a second time, which means a parent that kept failing would throw out + // of the handler that exists to contain it. + val actorService = new FailingParentActorService(workerId, newContext(parent)) + val service = new PekkoActorRefMappingService(actorService) + actorService.failuresLeft = 1 + + service.retrieveActorRef(destination, Set(waiter.ref)) + + // Nothing was asked and nobody was told: the lookup simply did not happen. + parent.expectNoMessage(100.millis) + waiter.expectNoMessage(100.millis) + + // ...and the id was not recorded as queried, so the next message bound for it re-asks. Marking + // it would strand every message for that destination: the reply that clears the stash only ever + // arrives in response to a `GetActorRef` that was actually sent. + service.retrieveActorRef(destination, Set(waiter.ref)) + + assert(parent.expectMsgType[GetActorRef].id == destination) + } } /** Minimal actor used only to obtain a live [[ActorContext]] from Pekko TestKit. */ class ActorRefMappingServiceContextHolder extends Actor { override def receive: Receive = { case _ => () } } + +/** A [[PekkoActorService]] whose first `failuresLeft` parent lookups throw. */ +class FailingParentActorService(id: ActorVirtualIdentity, actorContext: ActorContext) + extends PekkoActorService(id, actorContext) { + + var failuresLeft: Int = 0 + + override def parent: ActorRef = { + if (failuresLeft > 0) { + failuresLeft -= 1 + throw new IllegalStateException("parent is unreachable") + } + super.parent + } +} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/ConsoleMessageHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/ConsoleMessageHandlerSpec.scala new file mode 100644 index 0000000000..87d6cf744f --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/ConsoleMessageHandlerSpec.scala @@ -0,0 +1,134 @@ +/* + * 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.coordinator.promisehandlers + +import com.google.protobuf.timestamp.Timestamp +import com.twitter.util.{Await, Duration} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.core.workflow.WorkflowContext +import org.apache.texera.amber.engine.architecture.coordinator.{ + CoordinatorAsyncRPCHandlerInitializer, + CoordinatorConfig, + CoordinatorProcessor +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ConsoleMessage, + ConsoleMessageTriggeredRequest, + ConsoleMessageType +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.virtualidentity.util.{CLIENT, COORDINATOR} +import org.scalatest.flatspec.AnyFlatSpec + +import java.time.Instant +import scala.collection.mutable.ArrayBuffer + +/** + * `consoleMessageTriggered` is the coordinator's half of the console: a worker reports a print, an + * error or a debugger line (`DataProcessor.handleExecutorException`, and the Python runtime's + * equivalent), and the coordinator is the only hop between that worker and the browser. + * + * It is a pure relay, and everything that could break it is invisible from the handler itself: + * the message has to leave on the *client* channel (nothing else is subscribed to it, so a message + * sent anywhere else is simply lost), it has to arrive unmodified (the console panel groups rows + * by `workerId`/`msgType` and renders `title`/`message` verbatim), and it must not be duplicated + * onto a worker as well. + * + * `ConsoleMessage` is itself the `ClientEvent`, so this spec reads what the coordinator's output + * handler emits and asserts on the wire message the client actor would receive. The harness is the + * one from `EvaluatePythonExpressionHandlerSpec`; no ActorSystem is involved. + */ +class ConsoleMessageHandlerSpec extends AnyFlatSpec { + + private val awaitTimeout = Duration.fromSeconds(1) + + /** The worker that reported the message; also the RPC sender, as in production. */ + private val reportingWorkerId = ActorVirtualIdentity("Worker:WF1-udf-main-2") + private val rpcContext = AsyncRPCContext(reportingWorkerId, COORDINATOR) + + /** The channel the client actor listens on. */ + private val clientChannel = ChannelIdentity(COORDINATOR, CLIENT, isControl = true) + + /** + * Every text field holds a different value, so a relay that shuffled two of them (say `source` + * into `title`) cannot pass by accident. + */ + private val consoleMessage = ConsoleMessage( + workerId = reportingWorkerId.name, + timestamp = Timestamp(Instant.parse("2020-01-02T03:04:05Z")), + msgType = ConsoleMessageType.ERROR, + source = "(udf.py:31)", + title = "ZeroDivisionError: division by zero", + message = "Traceback (most recent call last)" + ) + + private def newFixture() + : (CoordinatorAsyncRPCHandlerInitializer, ArrayBuffer[WorkflowFIFOMessage]) = { + val sent = ArrayBuffer[WorkflowFIFOMessage]() + val outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit = { + case Right(m) => sent += m + case _ => () + } + val cp = new CoordinatorProcessor( + new WorkflowContext(), + CoordinatorConfig(None, None, None, None), + COORDINATOR, + outputHandler + ) + (new CoordinatorAsyncRPCHandlerInitializer(cp), sent) + } + + behavior of "ConsoleMessageHandler" + + it should "put the message on the client channel and nowhere else" in { + val (init, sent) = newFixture() + + init.consoleMessageTriggered(ConsoleMessageTriggeredRequest(consoleMessage), rpcContext) + + // Exactly one message: the console is a browser-facing feed, so relaying a copy back to a + // worker would be both useless and, for a paused worker, extra queued work. + assert(sent.size == 1) + assert(sent.head.channelId == clientChannel) + } + + it should "hand the client the reported message unchanged" in { + val (init, sent) = newFixture() + + init.consoleMessageTriggered(ConsoleMessageTriggeredRequest(consoleMessage), rpcContext) + + // The coordinator knows nothing about console formatting; the browser renders these fields + // directly, so the relayed payload has to be the reported one, field for field. + assert(sent.head.payload == consoleMessage) + } + + it should "acknowledge the reporting worker" in { + val (init, _) = newFixture() + + val response = + init.consoleMessageTriggered(ConsoleMessageTriggeredRequest(consoleMessage), rpcContext) + + // The worker treats this as a plain RPC and its promise stays unfulfilled until the reply + // comes back; the console is best-effort, so the reply cannot depend on the client. + assert(Await.result(response, awaitTimeout) == EmptyReturn()) + } +} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/DebugCommandHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/DebugCommandHandlerSpec.scala new file mode 100644 index 0000000000..933221ac21 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/DebugCommandHandlerSpec.scala @@ -0,0 +1,132 @@ +/* + * 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.coordinator.promisehandlers + +import com.twitter.util.{Await, Duration} +import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity +import org.apache.texera.amber.core.workflow.WorkflowContext +import org.apache.texera.amber.engine.architecture.coordinator.{ + CoordinatorAsyncRPCHandlerInitializer, + CoordinatorConfig, + CoordinatorProcessor +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ControlInvocation, + DebugCommandRequest +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.virtualidentity.util.{CLIENT, COORDINATOR} +import org.scalatest.flatspec.AnyFlatSpec + +import scala.collection.mutable.ArrayBuffer + +/** + * `debugCommand` is how a debugger command typed in the UI reaches one Python worker: the web + * layer calls it on the coordinator (`ExecutionConsoleService`), and the coordinator relays it to + * the single worker the request names. + * + * The routing is the whole handler, and it is easy to get wrong in a way nothing else notices: the + * target worker is carried in the request body, *not* in the RPC context, so a relay that reused + * `ctx` would send the command back to whoever asked (the client) instead of to the worker, and a + * breakpoint would silently never be set. + * + * The relay is also deliberately fire-and-forget: the reply is produced without waiting for the + * worker, because a worker stopped at a breakpoint answers only when it resumes and the UI must + * not block until then. + * + * The harness mirrors `EvaluatePythonExpressionHandlerSpec`: a real `CoordinatorProcessor` whose + * output handler collects the dispatched control messages, so what is asserted is the wire + * message a worker would receive. No ActorSystem and no workers are needed. + */ +class DebugCommandHandlerSpec extends AnyFlatSpec { + + private val awaitTimeout = Duration.fromSeconds(1) + + /** The worker named *in the request* — the one the command must reach. */ + private val targetWorkerId = ActorVirtualIdentity("Worker:WF1-udf-main-3") + + /** The command's sender, distinct from the target so a relay that reuses `ctx` is visible. */ + private val rpcContext = AsyncRPCContext(CLIENT, COORDINATOR) + + private val request = DebugCommandRequest(targetWorkerId.name, "break 12") + + private def newFixture() + : (CoordinatorAsyncRPCHandlerInitializer, ArrayBuffer[WorkflowFIFOMessage]) = { + val sent = ArrayBuffer[WorkflowFIFOMessage]() + val outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit = { + case Right(m) => sent += m + case _ => () + } + val cp = new CoordinatorProcessor( + new WorkflowContext(), + CoordinatorConfig(None, None, None, None), + COORDINATOR, + outputHandler + ) + (new CoordinatorAsyncRPCHandlerInitializer(cp), sent) + } + + /** The debug-command invocations the coordinator put on the wire. */ + private def dispatched(sent: ArrayBuffer[WorkflowFIFOMessage]): Seq[ControlInvocation] = + sent.toSeq.collect { + case WorkflowFIFOMessage(_, _, invocation: ControlInvocation) + if invocation.methodName == "debugCommand" => + invocation + } + + behavior of "DebugCommandHandler" + + it should "address the command to the worker the request names, not to the sender" in { + val (init, sent) = newFixture() + + init.debugCommand(request, rpcContext) + + val invocations = dispatched(sent) + assert(invocations.size == 1) + assert(invocations.head.context.receiver == targetWorkerId) + // The coordinator is the one asking the worker, so the relayed call is its own, not a + // forwarded copy of the client's context. + assert(invocations.head.context.sender == COORDINATOR) + } + + it should "relay the request body verbatim" in { + val (init, sent) = newFixture() + + init.debugCommand(request, rpcContext) + + // The worker-side handler parses `cmd` itself, so nothing may be rewritten, reordered or + // dropped on the way out. + assert(dispatched(sent).map(_.command) == Seq(request)) + } + + it should "answer without waiting for the worker to run the command" in { + val (init, sent) = newFixture() + + val response = init.debugCommand(request, rpcContext) + + // The worker's own reply is still outstanding — a worker sitting at a breakpoint answers only + // once it resumes — yet the coordinator has already answered its caller. + assert(dispatched(sent).size == 1) + assert(Await.result(response, awaitTimeout) == EmptyReturn()) + } +} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PrepareCheckpointHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PrepareCheckpointHandlerSpec.scala index 60e779ffa5..b375d97c3f 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PrepareCheckpointHandlerSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PrepareCheckpointHandlerSpec.scala @@ -20,30 +20,44 @@ package org.apache.texera.amber.engine.architecture.worker.promisehandlers import com.twitter.util.{Await, Duration, Future} +import org.apache.pekko.actor.{ActorSystem, Props} +import org.apache.pekko.testkit.{TestActorRef, TestKit} +import org.apache.texera.amber.clustering.SingleNodeListener import org.apache.texera.amber.core.executor.OperatorExecutor import org.apache.texera.amber.core.tuple.{Schema, Tuple, TupleLike} import org.apache.texera.amber.core.virtualidentity.{ ActorVirtualIdentity, + ChannelIdentity, EmbeddedControlMessageIdentity } import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ AsyncRPCContext, + EmptyRequest, PrepareCheckpointRequest } import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.rpc.workerservice.WorkerServiceGrpc.METHOD_QUERY_STATISTICS +import org.apache.texera.amber.engine.architecture.scheduling.config.WorkerConfig import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ DPInputQueueElement, - MainThreadDelegateMessage + FIFOMessageElement, + MainThreadDelegateMessage, + TimerBasedControlElement, + WorkerReplayInitialization } import org.apache.texera.amber.engine.architecture.worker.{ DataProcessor, - DataProcessorRPCHandlerInitializer + DataProcessorRPCHandlerInitializer, + WorkflowWorker } +import org.apache.texera.amber.engine.common.AmberRuntime import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.rpc.AsyncRPCClient.ControlInvocation import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.apache.texera.amber.engine.common.{CheckpointState, CheckpointSupport, SerializedState} -import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike import java.util.concurrent.LinkedBlockingQueue import scala.collection.mutable.ArrayBuffer @@ -65,20 +79,21 @@ import scala.collection.mutable.ArrayBuffer * iterator the operator returns — the un-emitted tuples would be lost on restore. * * The registered callback's last step hands a closure to the worker's main thread and blocks on - * it, which needs a live worker actor. These tests therefore install an output handler that throws - * [[PrepareCheckpointHandlerSpec.MainThreadHandOffReached]] at that hand-off: everything the - * callback does before it (the part that runs on the DP thread) is observable, and the assertions - * below stop there. + * it. Most of the cases below stop there: they install an output handler that throws + * [[PrepareCheckpointHandlerSpec.MainThreadHandOffReached]] at the hand-off, so everything the + * callback does before it (the part that runs on the DP thread) is observable in isolation. * - * Not covered here, and not covered anywhere else either: this handler's own main-thread closure, - * which drains the input queue into `DP_QUEUED_MSG_KEY`, snapshots un-acked output into - * `OUTPUT_MSG_KEY`, and starts input recording by seeding `worker.recordedInputs(checkpointId)`. - * `FinalizeCheckpointHandlerSpec` exercises a *different* closure and hand-installs - * `recordedInputs` itself, so the prepare-to-finalize handshake is currently unverified. Closing - * that gap needs a live worker actor (the `TestActorRef` recipe in `FinalizeCheckpointHandlerSpec` - * would do it) and is left as follow-up. + * The last case instead lets the hand-off through to a real worker actor, which is what covers the + * other half of the checkpoint: the main-thread closure drains the input queue into + * `DP_QUEUED_MSG_KEY`, snapshots un-acked output into `OUTPUT_MSG_KEY`, and starts input recording + * by seeding `worker.recordedInputs(checkpointId)` -- the buffer `finalizeCheckpoint` later folds + * in. `FinalizeCheckpointHandlerSpec` hand-installs that buffer, so this is the only place the + * prepare-to-finalize handshake is verified end to end. */ -class PrepareCheckpointHandlerSpec extends AnyFlatSpec { +class PrepareCheckpointHandlerSpec + extends TestKit(ActorSystem("PrepareCheckpointHandlerSpec", AmberRuntime.pekkoConfig)) + with AnyFlatSpecLike + with BeforeAndAfterAll { import PrepareCheckpointHandlerSpec._ @@ -87,6 +102,15 @@ class PrepareCheckpointHandlerSpec extends AnyFlatSpec { private val awaitTimeout = Duration.fromSeconds(5) private val checkpointId = EmbeddedControlMessageIdentity("prepare-checkpoint-1") + override def beforeAll(): Unit = { + // WorkflowActor's actor service resolves node addresses through /user/cluster-info. + system.actorOf(Props[SingleNodeListener](), "cluster-info") + } + + override def afterAll(): Unit = { + TestKit.shutdownActorSystem(system) + } + /** Records main-thread hand-offs, then aborts so the blocking wait is never reached. */ private class OutputRecorder { val delegates: ArrayBuffer[MainThreadDelegateMessage] = ArrayBuffer() @@ -115,6 +139,48 @@ class PrepareCheckpointHandlerSpec extends AnyFlatSpec { private def await[T](future: Future[T]): T = Await.result(future, awaitTimeout) + /** + * A live worker, so the main-thread hand-off is delivered the way production delivers it + * (`self !` -> `handleTriggerClosure`). `TestActorRef` dispatches on the calling thread, so the + * closure runs inside `dp.outputHandler` exactly as the blocking wait after it assumes. + */ + private def liveWorker(): TestActorRef[WorkflowWorker] = { + val worker = TestActorRef( + new WorkflowWorker(WorkerConfig(workerId), WorkerReplayInitialization()) + ) + // The DP thread would otherwise race the assertions on the worker's queue and recorded inputs. + worker.underlyingActor.dpThread.stop() + worker + } + + /** + * A handler whose DP shares the worker's input queue and routes main-thread hand-offs to it, + * which is the wiring `WorkflowWorker` builds for its own DP + * (`WorkflowActor.sendMessageFromLogWriterToActor`). + */ + private def newHandlerOn( + worker: TestActorRef[WorkflowWorker], + executor: OperatorExecutor + ): DataProcessorRPCHandlerInitializer = { + val dp = new DataProcessor( + workerId, + { + case Left(delegate) => worker ! delegate + case Right(_) => () + }, + worker.underlyingActor.inputQueue + ) + dp.executor = executor + new DataProcessorRPCHandlerInitializer(dp) + } + + private def queryStatisticsMessage(sequenceNumber: Long): WorkflowFIFOMessage = + WorkflowFIFOMessage( + ChannelIdentity(COORDINATOR, workerId, isControl = true), + sequenceNumber, + ControlInvocation(METHOD_QUERY_STATISTICS, EmptyRequest(), rpcContext, sequenceNumber) + ) + behavior of "PrepareCheckpointHandler" it should "register no serialization for an estimate-only request" in { @@ -194,6 +260,38 @@ class PrepareCheckpointHandlerSpec extends AnyFlatSpec { ) assert(handler.dp.ecmManager.checkpoints(checkpointId).has(OperatorStateKey)) } + + it should "save the worker's queued input and start recording what arrives next" in { + val worker = liveWorker() + val handler = newHandlerOn(worker, new PlainExecutor) + val queued = queryStatisticsMessage(7L) + worker.underlyingActor.inputQueue.put(FIFOMessageElement(queued)) + // Not a message from anywhere: the timer re-issues it after a restore, so checkpointing it + // would replay a statistics query the coordinator never sent. + worker.underlyingActor.inputQueue.put( + TimerBasedControlElement( + ControlInvocation(METHOD_QUERY_STATISTICS, EmptyRequest(), rpcContext, 8L) + ) + ) + + await(handler.prepareCheckpoint(PrepareCheckpointRequest(checkpointId, false), rpcContext)) + handler.dp.serializationManager.applySerialization() + + val checkpoint = handler.dp.ecmManager.checkpoints(checkpointId) + // Messages that arrived but have not been processed are part of the worker's state: on restore + // they go back into the queue, so anything dropped here is a lost control message or batch. + assert( + checkpoint + .load[ArrayBuffer[WorkflowFIFOMessage]](SerializedState.DP_QUEUED_MSG_KEY) + .toList == List(queued) + ) + // Nothing was sent, so nothing is awaiting an ack -- but the key still has to exist, because + // `WorkflowWorker.loadFromCheckpoint` reads it unconditionally. + assert(checkpoint.load[Array[WorkflowFIFOMessage]](SerializedState.OUTPUT_MSG_KEY).isEmpty) + // The hand-off ends by opening this buffer, and `finalizeCheckpoint` folds it in: without it, + // every message arriving between the two halves of the checkpoint would be lost on restore. + assert(worker.underlyingActor.recordedInputs.keySet == Set(checkpointId)) + } } object PrepareCheckpointHandlerSpec { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandlerSpec.scala new file mode 100644 index 0000000000..813aeacdc7 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandlerSpec.scala @@ -0,0 +1,128 @@ +/* + * 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.promisehandlers + +import com.twitter.util.{Await, Duration} +import org.apache.texera.amber.core.WorkflowRuntimeException +import org.apache.texera.amber.core.executor.OperatorExecutor +import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + EmptyRequest +} +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.{ + DPInputQueueElement, + MainThreadDelegateMessage +} +import org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState.UNINITIALIZED +import org.apache.texera.amber.engine.architecture.worker.{ + DataProcessor, + DataProcessorRPCHandlerInitializer +} +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR +import org.scalatest.flatspec.AnyFlatSpec + +import java.util.concurrent.LinkedBlockingQueue +import scala.collection.mutable.ArrayBuffer + +/** + * `startWorker` is sent only to the workers that begin a region: a source operator, which produces + * its own input, and an operator whose input ports are fed from materialized storage rather than + * from an upstream worker. Every other worker starts because data arrives on one of its channels, + * and receiving `StartWorker` means the scheduler picked the wrong worker set. + * + * That third case is the one pinned here. It is a hard failure on purpose: a worker that quietly + * accepted the message would transition to RUNNING with no input to process and no channel that + * will ever finish, and the region would hang instead of reporting a scheduling bug. The two + * starting cases are covered where the workers really run; what has no coverage anywhere is the + * refusal, and a refusal that stopped firing would be invisible until a workflow hung. + * + * The harness is the bare `DataProcessor` of the other worker handler specs — no ActorSystem, and + * no operator ever runs, because the handler must reject before doing any work. + */ +class StartHandlerSpec extends AnyFlatSpec { + + import StartHandlerSpec._ + + private val workerId = ActorVirtualIdentity("Worker:WF1-filter-main-2") + private val rpcContext = AsyncRPCContext(COORDINATOR, workerId) + private val awaitTimeout = Duration.fromSeconds(5) + + /** + * A worker of a mid-workflow operator: not a source, and with no input port bound to + * materialized storage, so it has no way to start itself. + */ + private def newHandler() + : (DataProcessorRPCHandlerInitializer, ArrayBuffer[WorkflowFIFOMessage]) = { + val sent = ArrayBuffer[WorkflowFIFOMessage]() + val outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit = { + case Right(m) => sent += m + case _ => () + } + val dp = new DataProcessor( + workerId, + outputHandler, + new LinkedBlockingQueue[DPInputQueueElement]() + ) + dp.executor = new NonSourceExecutor + (new DataProcessorRPCHandlerInitializer(dp), sent) + } + + behavior of "StartHandler" + + it should "refuse a worker that is neither a source nor a reader of materialized input" in { + val (handler, _) = newHandler() + + val failure = intercept[WorkflowRuntimeException] { + Await.result(handler.startWorker(EmptyRequest(), rpcContext), awaitTimeout) + } + + // The message has to name the worker: the coordinator logs it as a control-message failure, and + // the worker set it came from is the only clue to which operator the scheduler mis-selected. + assert(failure.getMessage.contains(workerId.name)) + // Distinguishes this refusal from `WorkerStateManager`'s own `InvalidStateException`, which is + // also a `WorkflowRuntimeException` and is what an accepted-but-not-READY worker would raise. + assert(failure.getMessage.contains("unexpected StartWorker")) + } + + it should "refuse before touching the worker" in { + val (handler, sent) = newHandler() + + intercept[WorkflowRuntimeException] { + Await.result(handler.startWorker(EmptyRequest(), rpcContext), awaitTimeout) + } + + // A refusal that had already half-started the worker would leave it advertising a state it + // cannot back out of, and would have told the downstream region its channel had opened. + assert(handler.dp.stateManager.getCurrentState == UNINITIALIZED) + assert(handler.dp.inputManager.getAllPorts.isEmpty) + assert(sent.isEmpty) + } +} + +object StartHandlerSpec { + + /** Not a `SourceOperatorExecutor`, so the handler's source branch cannot apply. */ + class NonSourceExecutor extends OperatorExecutor { + override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = Iterator.empty + } +} diff --git a/amber/src/test/scala/org/apache/texera/web/resource/SystemMetadataResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/SystemMetadataResourceSpec.scala new file mode 100644 index 0000000000..028de9ae7c --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/web/resource/SystemMetadataResourceSpec.scala @@ -0,0 +1,66 @@ +/* + * 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.web.resource + +import org.apache.texera.amber.operator.metadata.{ + AllOperatorMetadata, + OperatorGroupConstants, + OperatorMetadataGenerator +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** + * `GET /api/resources/operator-metadata` is the first call the workspace makes: the palette, the + * property editor and the link validator are all built from this one payload, so an endpoint that + * answered with a trimmed or rebuilt catalogue would leave the editor with operators it cannot + * place or forms it cannot render. + * + * The resource only delegates, and the delegation is what these tests pin — that a caller receives + * the generated catalogue whole, operators and palette groups alike. How the catalogue itself is + * derived (JSON-schema generation, port declarations) belongs to `OperatorMetadataGenerator` and is + * not re-asserted here. + * + * `TexeraWebApplicationSpec` separately asserts that this resource is registered on Jersey, but + * nothing there calls it, so until now nothing has run the method behind the endpoint. + */ +class SystemMetadataResourceSpec extends AnyFlatSpec with Matchers { + + private val response: AllOperatorMetadata = new SystemMetadataResource().getOperatorMetadata + + behavior of "SystemMetadataResource" + + it should "hand back an entry for every operator the workspace can place" in { + // Every `LogicalOp` subtype registered for JSON polymorphism is an operator a saved workflow may + // reference, so one missing entry is an operator the editor cannot render. + response.operators.map(_.operatorType) should contain theSameElementsAs + OperatorMetadataGenerator.operatorTypeMap.values.toList + // The comparison above says the two agree, not that either holds anything: these two operators + // are always registered, so they also rule out an agreeing-but-empty catalogue. + response.operators.map(_.operatorType) should contain allOf ("CSVFileScan", "Filter") + } + + it should "hand back the palette's group order" in { + // The palette renders its sections in this order. The list is declared, not derived from the + // operators, so the endpoint has to pass it through rather than reconstruct it. + response.groups shouldBe OperatorGroupConstants.OperatorGroupOrderList + response.groups should not be empty + } +}
