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-7441-12eaccbb24884fccf1dba26d15fff7ceb1db4aba in repository https://gitbox.apache.org/repos/asf/texera.git
commit 08a2eac8eb7151eaac33b18cd8a47a20599e5fa7 Author: Xinyuan Lin <[email protected]> AuthorDate: Sun Aug 9 16:49:50 2026 -0700 test(amber): cover ExecutionConsoleService's console routing and debug commands (#7441) ### What changes were proposed in this PR? The suite covered the `ConsoleMessageProcessor` object; the service class around it was untouched. That class owns the console diff the frontend is driven from, the worker-to-operator keying that decides where a message lands, and the websocket handler behind the debugger. Converts the spec to a TestKit suite and adds 8 tests: - a **debugger** message is never truncated, while an ordinary one still is — the debugger's output is the frame the user asked to see - the diff reports **only messages added since the last state**; the frontend appends what it is sent, so emitting the whole buffer would duplicate every earlier line on each update - a console message is filed under the **logical** operator id — the worker id carries the physical layer and worker index, and anything else strands the output where the frontend will not look for it - a debug command is attributed to `USER-<uid>`, falls back to `USER-UNKNOWN` with no session user, is filed under the operator rather than the worker, and is forwarded to the coordinator with the worker it names **Verified by mutation**, all reverted (production diff empty): | Mutation | Result | |---|---| | truncate debugger messages too | red | | send the whole console buffer instead of the diff | red | | file the message under the physical op id | red | | file the message under the raw worker id | red | | attribute every debug command to a constant | red | | file the debug command under the worker | red | | forward the wrong id to the coordinator | red | | drop the command from the message title | red | Everything runs on an empty-plan `AmberClient` with a mocked coordinator: no engine, database or Iceberg storage. The Iceberg-backed writer path (`getOrCreateWriter` and the execution-state commit loop) is `private` and storage-bound; it is left uncovered rather than padded with a no-throw test, and the spec says so. One note recorded in the spec: the console store's event observable replays on subscribe, so the diff test subscribes first and asserts on the batch published for the second message. No production file is touched. ### Any related issues, documentation, discussions? Closes #7438 ### How was this PR tested? ``` sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionConsoleServiceSpec" ``` ``` [info] Tests: succeeded 13, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. ``` 8 new on top of the existing 5. `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) --------- Signed-off-by: Xinyuan Lin <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../web/service/ExecutionConsoleServiceSpec.scala | 218 ++++++++++++++++++++- 1 file changed, 214 insertions(+), 4 deletions(-) diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala index d4753984cf..b1d647d035 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionConsoleServiceSpec.scala @@ -20,17 +20,57 @@ package org.apache.texera.web.service import com.google.protobuf.timestamp.Timestamp +import com.twitter.util.{Future => TwitterFuture} +import io.reactivex.rxjava3.disposables.Disposable +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.TestKit +import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext} +import org.apache.texera.amber.engine.architecture.coordinator.CoordinatorConfig import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ ConsoleMessage, - ConsoleMessageType + ConsoleMessageType, + DebugCommandRequest => AmberDebugCommandRequest } +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceFs2Grpc +import org.apache.texera.amber.engine.common.client.AmberClient import org.apache.texera.amber.engine.common.executionruntimestate.ExecutionConsoleStore -import org.scalatest.flatspec.AnyFlatSpec +import org.apache.texera.web.WebsocketInput +import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent +import org.apache.texera.web.model.websocket.event.python.ConsoleUpdateEvent +import org.apache.texera.web.model.websocket.request.python.DebugCommandRequest +import org.apache.texera.web.storage.ExecutionStateStore +import org.scalamock.scalatest.MockFactory +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike import org.scalatest.matchers.should.Matchers import java.time.Instant - -class ExecutionConsoleServiceSpec extends AnyFlatSpec with Matchers { +import scala.collection.mutable.ListBuffer +import scala.reflect.ClassTag + +/** + * The `ConsoleMessageProcessor` object is covered by the first half of this suite. The service + * class around it was entirely uncovered: it owns the console diff that decides what the frontend + * is told, the worker-to-operator keying that decides where a message lands, and the websocket + * handlers behind the debugger. + * + * Everything here runs on an empty-plan AmberClient with a mocked coordinator, so no engine is + * involved. Note that ExecutionConsoleService still schedules asynchronous console persistence + * (writer creation + operator-executions insertion); this suite focuses on console routing/diffing + * and does not assert on the persistence side effects. + */ +class ExecutionConsoleServiceSpec + extends TestKit(ActorSystem("ExecutionConsoleServiceSpec")) + with AnyFlatSpecLike + with Matchers + with MockFactory + with BeforeAndAfterAll { + + override def afterAll(): Unit = { + try TestKit.shutdownActorSystem(system) + finally super.afterAll() + } // Constants for testing val standardBufferSize: Int = 100 @@ -231,4 +271,174 @@ class ExecutionConsoleServiceSpec extends AnyFlatSpec with Matchers { val expectedTruncatedTitle = "a" * (messageDisplayLength - 3) + "..." opInfo.consoleMessages.head.title shouldBe expectedTruncatedTitle } + // ---------------------------------------------------------------- instance + + /** Empty-plan client that captures the ConsoleMessage callback the service registers. */ + private final class TestAmberClient( + override val coordinatorInterface: CoordinatorServiceFs2Grpc[TwitterFuture, Unit] + ) extends AmberClient( + system, + new WorkflowContext(), + PhysicalPlan(Set.empty, Set.empty), + CoordinatorConfig(None, None, None, None), + _ => () + ) { + var consoleCallback: ConsoleMessage => Unit = _ + + override def registerCallback[T](callback: T => Unit)(implicit ct: ClassTag[T]): Disposable = { + if (ct.runtimeClass == classOf[ConsoleMessage]) { + consoleCallback = callback.asInstanceOf[ConsoleMessage => Unit] + } + Disposable.empty() + } + + def dispose(): Unit = super.shutdown() + } + + private final class Fixture( + val client: TestAmberClient, + val coordinator: CoordinatorServiceFs2Grpc[TwitterFuture, Unit], + val stateStore: ExecutionStateStore, + val wsInput: WebsocketInput, + val service: ExecutionConsoleService + ) { + def close(): Unit = { + service.unsubscribeAll() + client.dispose() + } + } + + private def fixture(): Fixture = { + val coordinator = mock[CoordinatorServiceFs2Grpc[TwitterFuture, Unit]] + val client = new TestAmberClient(coordinator) + val stateStore = new ExecutionStateStore + val wsInput = new WebsocketInput(ListBuffer.empty[Throwable] += _) + val service = new ExecutionConsoleService(client, stateStore, wsInput, new WorkflowContext()) + new Fixture(client, coordinator, stateStore, wsInput, service) + } + + private def message( + workerId: String = "Worker:WF1-udf1-main-0", + title: String = "hello", + msgType: ConsoleMessageType = ConsoleMessageType.PRINT + ): ConsoleMessage = + new ConsoleMessage(workerId, Timestamp(Instant.now), msgType, "src", title, "content") + + private def withFixture(body: Fixture => Unit): Unit = { + val f = fixture() + try body(f) + finally f.close() + } + + "processConsoleMessage" should "leave a debugger message untouched however long it is" in { + // The debugger's output is the payload the user asked to see; truncating it would cut off the + // frame or variable they are inspecting. + withFixture { f => + val long = "a" * (f.service.consoleMessageDisplayLength + 50) + + val processed = f.service.processConsoleMessage( + message(title = long, msgType = ConsoleMessageType.DEBUGGER) + ) + + processed.title shouldBe long + } + } + + it should "still truncate an ordinary message to the configured length" in { + withFixture { f => + val long = "a" * (f.service.consoleMessageDisplayLength + 50) + + val processed = f.service.processConsoleMessage(message(title = long)) + + processed.title.length shouldBe f.service.consoleMessageDisplayLength + processed.title should endWith("...") + } + } + + "the console diff handler" should "report only the messages added since the last state" in { + // The frontend appends what it is sent. Emitting the whole buffer instead of the delta would + // duplicate every earlier line on each update. + withFixture { f => + // One batch is published per state update. Subscribing up front and reading the batch for the + // SECOND message is what shows the delta: the first message must not appear in it again. + val batches = ListBuffer.empty[Iterable[TexeraWebSocketEvent]] + val sub = f.stateStore.consoleStore.getWebsocketEventObservable + .subscribe((batch: Iterable[TexeraWebSocketEvent]) => batches += batch) + + try { + f.client.consoleCallback(message(title = "first")) + f.client.consoleCallback(message(title = "second")) + } finally sub.dispose() + + val titles = + batches.last.collect { case e: ConsoleUpdateEvent => e.messages.map(_.title) }.flatten + titles shouldBe Seq("second") + } + } + + "the console message callback" should "file a message under the logical operator id" in { + // The worker id carries the physical layer and worker index; the frontend console is keyed by + // the logical operator, so anything else silently strands the output. + withFixture { f => + f.client.consoleCallback(message(workerId = "Worker:WF1-udf1-main-0")) + + f.stateStore.consoleStore.getState.operatorConsole.keys should contain("udf1") + } + } + + it should "store the truncated form, not the original" in { + withFixture { f => + val long = "b" * (f.service.consoleMessageDisplayLength + 50) + + f.client.consoleCallback(message(title = long)) + + val stored = f.stateStore.consoleStore.getState.operatorConsole("udf1").consoleMessages + stored.map(_.title.length) shouldBe Seq(f.service.consoleMessageDisplayLength) + } + } + + "a debug command" should "be attributed to the user that issued it" in { + withFixture { f => + (f.coordinator.debugCommand _) + .expects(AmberDebugCommandRequest("Worker:WF1-udf1-main-0", "break 12"), ()) + .returning(TwitterFuture.value(EmptyReturn())) + + f.wsInput.onNext( + DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "break 12"), + Some(7) + ) + + val stored = f.stateStore.consoleStore.getState.operatorConsole("udf1").consoleMessages + stored.map(_.source) shouldBe Seq("USER-7") + stored.map(_.title) shouldBe Seq("break 12") + } + } + + it should "fall back to UNKNOWN when there is no session user" in { + withFixture { f => + (f.coordinator.debugCommand _) + .expects(*, *) + .returning(TwitterFuture.value(EmptyReturn())) + + f.wsInput.onNext(DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "cont"), None) + + f.stateStore.consoleStore.getState + .operatorConsole("udf1") + .consoleMessages + .map(_.source) shouldBe Seq("USER-UNKNOWN") + } + } + + it should "file the command under the operator, not the worker" in { + // req carries both; keying by workerId would scatter the command across per-worker consoles. + withFixture { f => + (f.coordinator.debugCommand _).expects(*, *).returning(TwitterFuture.value(EmptyReturn())) + + f.wsInput.onNext(DebugCommandRequest("udf1", "Worker:WF1-udf1-main-0", "cont"), Some(1)) + + val keys = f.stateStore.consoleStore.getState.operatorConsole.keys + keys should contain("udf1") + keys should not contain "Worker:WF1-udf1-main-0" + } + } }
