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-7648-363537e0251182472f540079d262599f6cbb5240 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 9f6d373ad5e6e7eaef3a15d0337b23d16796bd71 Author: Eugene Gu <[email protected]> AuthorDate: Sun Aug 16 08:13:56 2026 +0000 test(amber): cover AdvanceRegionExecutionsHandler (#7648) ### What changes were proposed in this PR? This adds `AdvanceRegionExecutionsHandlerSpec`, the first direct coverage of `AdvanceRegionExecutionsHandler` (added in #6960). The handler is four lines long but carries three distinct contracts, none of which was asserted anywhere: it delegates the advance to the workflow execution manager with the coordinator's own actor service, it replies to the requesting round *without* awaiting that advance (the whole point of routing the advance through its own control round), and on failure it notifies the client with a `FatalError` that carries the related worker id when the cause is a `WorkflowRuntimeException` and `None` otherwise. Six tests, test-only — no production file is touched: 1. `should advance the region executions with the coordinator's own actor service` — asserts the forwarded `PekkoActorService` is the coordinator's own instance (reference equality, plus a non-null guard so a vacuous `null == null` cannot pass). 2. `should reply immediately instead of awaiting the advance` — the reply is defined while the advance's promise is still unsatisfied. Deterministic, not timing-based: the handler's `EmptyReturn()` becomes an already-satisfied future through `returnAsFuture`, while the stubbed advance never completes. No sleeps or polling anywhere in the spec. 3. `should reply to the requesting round while the advance is still running` — delivers the request through `processDCM` and the real `AsyncRPCServer`, so it also pins that the handler is dispatchable under `METHOD_COORDINATOR_INITIATE_ADVANCE_REGION_EXECUTIONS`, the wire name `PortCompletedHandler` actually sends. 4. `should notify the client of a failed advance with the related worker id` 5. `should notify the client of a failed advance without a worker id for other failures` 6. `should not notify the client when the advance succeeds` The fixture pattern is reused from the sibling handler specs (`TestKit` + `RegionExecutionManagerTestSupport.createCoordinatorHarness()` + a real `CoordinatorProcessor` and `CoordinatorAsyncRPCHandlerInitializer`, as in `PortCompletedHandlerSpec` and `EmbeddedControlMessageHandlerSpec`). The only new piece is a `WorkflowExecutionManager` subclass that records advance calls and hands back a promise the test controls: driving the real manager into an async failure requires staging a full region launch/termination, and the causes it produces there are `IllegalStateException`s or the bare `Throwable` from `ErrorUtils.reconstructThrowable` — never a `WorkflowRuntimeException` carrying a worker id, which is the case test 4 exists to pin. Only the advance is stubbed; the coordinator processor, RPC server dispatch, RPC client and output gateway are all real. Note on test 4: the `WorkflowRuntimeException` arm currently pins *intent* rather than today-observable behavior. The two production subclasses that do carry a worker id, `StateManager.InvalidStateException` and `InvalidTransitionException` (`StateManager.scala:31-35`), are thrown worker-side, and a failed worker RPC reaches the coordinator through `ErrorUtils.reconstructThrowable`, which always rebuilds a plain `Throwable` and so drops the type; every other production site constructs a `WorkflowRuntimeException` with a message only, leaving `relatedWorkerId` empty. Both arms would therefore behave identically at present. The branch is in the handler and documented as intended, so the test guards it against regression. ### Any related issues, documentation, discussions? Closes #7646 The handler was introduced by #6960 (backported as #7096). ### How was this PR tested? `sbt "WorkflowExecutionService/testOnly *AdvanceRegionExecutionsHandlerSpec"` — 6 tests, all passing. No regressions in the neighbourhood: `*coordinator.promisehandlers.* *scheduling.*` runs 20 suites / 166 tests / 0 failures. `WorkflowExecutionService/Test/scalafmtCheck` is clean. Each test was mutation-checked: the production handler was temporarily broken, the suite re-run, and the handler reverted. Every behavior-changing mutant is caught, and each test is the sole killer of at least one: | Mutant | Caught by | |---|---| | `FatalError(err, err.relatedWorkerId)` → `FatalError(err, None)` | 4 | | `FatalError(err, Some(COORDINATOR))` — wrong but non-empty worker id | 4 | | drop the `.onFailure` block | 4, 5 | | `.onFailure { case _ => () }` (swallow the failure) | 4, 5 | | collapse both arms into a single catch-all | 4 | | notify on `.onSuccess` instead | 4, 5, 6 | | send the `FatalError` twice | 4 | | await the advance (`.map(_ => EmptyReturn())`) | 2, 3 | | never delegate (`Future.Unit` in place of the call) | 1, 2, 3, 4, 5 | | `advanceRegionExecutions(null)` instead of `cp.actorService` | 1 | | call `advanceRegionExecutions` twice | 3 | The only surviving mutant, `EmptyReturn()` → `Future.value(EmptyReturn())`, is semantically equivalent (`returnAsFuture` is an eager `Future.apply`), so there is nothing to catch. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) Co-authored-by: Meng Wang <[email protected]> Co-authored-by: Xinyuan Lin <[email protected]> --- .../AdvanceRegionExecutionsHandlerSpec.scala | 247 +++++++++++++++++++++ 1 file changed, 247 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/AdvanceRegionExecutionsHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/AdvanceRegionExecutionsHandlerSpec.scala new file mode 100644 index 0000000000..a942c615aa --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/AdvanceRegionExecutionsHandlerSpec.scala @@ -0,0 +1,247 @@ +/* + * 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, Future, Promise} +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.TestKit +import org.apache.texera.amber.core.WorkflowRuntimeException +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.core.workflow.WorkflowContext +import org.apache.texera.amber.engine.architecture.common.PekkoActorService +import org.apache.texera.amber.engine.architecture.coordinator.execution.WorkflowExecution +import org.apache.texera.amber.engine.architecture.coordinator.{ + ClientEvent, + CoordinatorAsyncRPCHandlerInitializer, + CoordinatorConfig, + CoordinatorProcessor, + FatalError +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + EmptyRequest +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{ + EmptyReturn, + ReturnInvocation +} +import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceGrpc.METHOD_COORDINATOR_INITIATE_ADVANCE_REGION_EXECUTIONS +import org.apache.texera.amber.engine.architecture.scheduling.{ + RegionExecutionManagerTestSupport, + WorkflowExecutionManager +} +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage +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 +import org.apache.texera.amber.engine.common.rpc.AsyncRPCClient.ControlInvocation +import org.apache.texera.amber.engine.common.virtualidentity.util.{CLIENT, COORDINATOR} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike + +import scala.collection.mutable.ArrayBuffer + +/** + * `coordinatorInitiateAdvanceRegionExecutions` is how the coordinator advances its region + * executions in a control round of its own: a handler that must not advance inline (see + * `PortCompletedHandler`) sends this request to the coordinator itself, so that the `EndWorker` + * messages a completed region produces cannot overtake the replies that round still owed. + * + * The behaviors this spec pins down: + * - the advance is delegated to the coordinator's own `WorkflowExecutionManager`; + * - the reply is produced synchronously, WITHOUT awaiting the advance. The requester is the + * coordinator itself and discards the reply, so awaiting it would only hold the control round + * open; + * - because nothing awaits the reply, a failing advance has no caller to propagate to and is + * reported to the client as a `FatalError` instead — carrying the related worker id when the + * failure came from a worker RPC (`WorkflowRuntimeException`) and `None` for anything else. + */ +class AdvanceRegionExecutionsHandlerSpec + extends TestKit(ActorSystem("AdvanceRegionExecutionsHandlerSpec", AmberRuntime.pekkoConfig)) + with AnyFlatSpecLike + with BeforeAndAfterAll + with RegionExecutionManagerTestSupport { + + private val awaitTimeout = Duration.fromSeconds(5) + private val coordinatorConfig = CoordinatorConfig(None, None, None, None) + private val ctx: AsyncRPCContext = AsyncRPCContext(COORDINATOR, COORDINATOR) + private val advanceCommandId = 11L + private val relatedWorker = ActorVirtualIdentity("Worker:unschedulable-region-worker") + + override def afterAll(): Unit = { + TestKit.shutdownActorSystem(system) + } + + /** + * Stands in for the real `WorkflowExecutionManager`: driving its advance to a *failed* future + * from the outside means staging a whole region launch or termination, and the causes that + * reaches the handler with are `IllegalStateException`s (unresolved output schema, termination + * retries exhausted) or the bare `Throwable` that `ErrorUtils.reconstructThrowable` builds for a + * worker RPC error — never a `WorkflowRuntimeException` carrying a worker id. This stub records + * the advance calls and hands back a future the test controls, so the handler can be observed + * both while the advance is still pending and after it has failed with either kind of cause. + */ + private class PendingAdvanceExecutionManager( + workflowExecution: WorkflowExecution, + asyncRPCClient: AsyncRPCClient + ) extends WorkflowExecutionManager(workflowExecution, coordinatorConfig, asyncRPCClient) { + val advanceCalls: ArrayBuffer[PekkoActorService] = ArrayBuffer() + private val advance: Promise[Unit] = Promise[Unit]() + + override def advanceRegionExecutions(actorService: PekkoActorService): Future[Unit] = { + advanceCalls += actorService + advance + } + + /** Twitter futures run their continuations on the thread that satisfies them, so the + * handler's `onFailure` has already run once this returns — no polling needed. + */ + def failAdvance(cause: Throwable): Unit = advance.setException(cause) + + def finishAdvance(): Unit = advance.setValue(()) + } + + private class TestCoordinatorProcessor( + outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit + ) extends CoordinatorProcessor( + new WorkflowContext(), + coordinatorConfig, + COORDINATOR, + outputHandler + ) { + override val workflowExecutionManager: PendingAdvanceExecutionManager = + new PendingAdvanceExecutionManager(workflowExecution, asyncRPCClient) + } + + private case class Fixture( + cp: TestCoordinatorProcessor, + init: CoordinatorAsyncRPCHandlerInitializer, + sent: ArrayBuffer[WorkflowFIFOMessage] + ) { + def manager: PendingAdvanceExecutionManager = cp.workflowExecutionManager + } + + /** + * A real `CoordinatorProcessor` (and therefore the real RPC layer) with the execution manager + * replaced by the controllable stub, and a real `PekkoActorService` set up so the argument the + * handler forwards is something other than `null`. + */ + private def newFixture(): Fixture = { + val sent = ArrayBuffer[WorkflowFIFOMessage]() + val cp = new TestCoordinatorProcessor({ + case Right(msg) => sent += msg + case Left(_) => () + }) + cp.setupActorService(createCoordinatorHarness().actorService) + Fixture(cp, new CoordinatorAsyncRPCHandlerInitializer(cp), sent) + } + + private def clientEvents(sent: ArrayBuffer[WorkflowFIFOMessage]): Seq[ClientEvent] = + sent.toSeq.filter(_.channelId.toWorkerId == CLIENT).map(_.payload).collect { + case event: ClientEvent => event + } + + private def repliesToCoordinator(sent: ArrayBuffer[WorkflowFIFOMessage]): Seq[ReturnInvocation] = + sent.toSeq.filter(_.channelId.toWorkerId == COORDINATOR).map(_.payload).collect { + case ret: ReturnInvocation => ret + } + + behavior of "AdvanceRegionExecutionsHandler" + + it should "advance the region executions with the coordinator's own actor service" in { + val fixture = newFixture() + + fixture.init.coordinatorInitiateAdvanceRegionExecutions(EmptyRequest(), ctx) + + // The actor service is what the advance needs to create the next region's workers on, so the + // coordinator's own one must be forwarded; guarded against a vacuous null == null match. + assert(fixture.cp.actorService != null) + assert(fixture.manager.advanceCalls.toSeq == Seq(fixture.cp.actorService)) + } + + it should "reply immediately instead of awaiting the advance" in { + val fixture = newFixture() + + val response = fixture.init.coordinatorInitiateAdvanceRegionExecutions(EmptyRequest(), ctx) + + // The advance was started but is still pending, and the reply is already satisfied: this + // handler deliberately does not chain its reply on the advance. Chaining it would keep the + // control round open for the whole advance, which itself waits on region termination RPCs. + assert(fixture.manager.advanceCalls.size == 1) + assert(response.isDefined) + assert(Await.result(response, awaitTimeout) == EmptyReturn()) + } + + it should "reply to the requesting round while the advance is still running" in { + val fixture = newFixture() + + // Delivered the way production does it: as a coordinator-to-coordinator control message, + // dispatched by the real RPC server. No other spec drives this handler that way — + // `PortCompletedHandlerSpec` only asserts that the invocation is sent. + fixture.cp.processDCM( + ChannelIdentity(COORDINATOR, COORDINATOR, isControl = true), + ControlInvocation( + METHOD_COORDINATOR_INITIATE_ADVANCE_REGION_EXECUTIONS, + EmptyRequest(), + ctx, + advanceCommandId + ) + ) + + assert(fixture.manager.advanceCalls.size == 1) + assert( + repliesToCoordinator(fixture.sent) == Seq( + ReturnInvocation(advanceCommandId, EmptyReturn()) + ) + ) + } + + it should "notify the client of a failed advance with the related worker id" in { + val fixture = newFixture() + fixture.init.coordinatorInitiateAdvanceRegionExecutions(EmptyRequest(), ctx) + + val failure = new WorkflowRuntimeException("region cannot be scheduled", Some(relatedWorker)) + fixture.manager.failAdvance(failure) + + // A `WorkflowRuntimeException` from a worker RPC knows which worker it came from, and the + // client needs that id to attribute the error to an operator. + assert(clientEvents(fixture.sent) == Seq(FatalError(failure, Some(relatedWorker)))) + } + + it should "notify the client of a failed advance without a worker id for other failures" in { + val fixture = newFixture() + fixture.init.coordinatorInitiateAdvanceRegionExecutions(EmptyRequest(), ctx) + + val failure = new IllegalStateException("no resource config for the next region") + fixture.manager.failAdvance(failure) + + // Not every advance failure comes from a worker, so there is no id to attribute. + assert(clientEvents(fixture.sent) == Seq(FatalError(failure, None))) + } + + it should "not notify the client when the advance succeeds" in { + val fixture = newFixture() + fixture.init.coordinatorInitiateAdvanceRegionExecutions(EmptyRequest(), ctx) + + fixture.manager.finishAdvance() + + assert(clientEvents(fixture.sent).isEmpty) + } +}
