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-7802-61cb36eab900b694bc00c45e5dd8eebb47266d82 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 53ad65021b875e025d23b643f5f02f560afc8492 Author: Xinyuan Lin <[email protected]> AuthorDate: Thu Sep 24 03:42:49 2026 +0000 fix(amber): report an uninitialized execution instead of NPEing, and map unparseable frames (#7802) ### What changes were proposed in this PR? Three defects in `WorkflowWebsocketResource`, all previously characterized without being cemented (#7303, #7676) so a fix would not have to fight a test asserting the broken behaviour. **1 & 2. A workflow with no execution NPEs instead of reporting "not initialized".** `WorkflowService.executionService` is a `BehaviorSubject` with **no initial value** (`WorkflowService.scala:140`), so `getValue` is `null` until an execution is published. - `case other =>` used `workflowStateOpt.map(_.executionService.getValue)`, which wraps that null into `Some(null)` — walking past the `case None` arm that exists to report the friendly error, then NPEing on `value.wsInput`. Now uses the already-computed `executionStateOpt`, which is built with `Option(...)`, so `Some(null)` cannot form and the existing `None` arm actually fires. - The `ModifyLogicRequest` arm had the same gap in a different shape: its guard tested the *workflow* where it meant the *execution*. Now `executionStateOpt.getOrElse(throw new IllegalStateException("workflow execution is not initialized"))`. Line 89 of this file already used `Option(...)` correctly, as does `WorkflowService` at its lines 208 and 349 — the fix adopts the established in-tree idiom rather than inventing one. **A shape decision worth reviewing.** For the `ModifyLogicRequest` arm I did *not* simply swap the outer condition to `executionStateOpt.isDefined`. That variant makes a workflow-without-execution **silently do nothing** instead of reporting, which is not the intent — and it would collide with the pre-existing test "ignore a ModifyLogicRequest that arrives before any workflow is attached", which pins the no-workflow case as `noException` plus `sent shouldBe empty`. Keeping the workflow guard and reporting the absent execution satisfies both. **3. An unparseable frame no longer escapes the error mapper.** `objectMapper.readValue` moved from above the `try` to its first statement, so a frame the mapper cannot bind is reported like any handler failure. `sessionState` and `executionStateOpt` stay outside, because the `catch` arm needs them — both routing arms still work. All messages use the existing wording, `"workflow execution is not initialized"`. ### The fixes are pinned Four new tests. Verified in both directions, with the production file reverted and restored: | | production reverted | with fixes | |---|---|---| | `WorkflowWebsocketResourceSpec` | **15 passed, 4 failed** | **19 passed, 0 failed** | The before-state failures are the right ones, from the JUnit XML (sbt's only reporter here is `-u`, so the console shows no per-test lines): | new test | failure without the fix | |---|---| | runtime command with a workflow but no execution | `Expected java.lang.IllegalStateException … java.lang.NullPointerException was thrown` | | `ModifyLogicRequest` before any execution exists | `Expected java.lang.IllegalStateException … java.lang.NullPointerException was thrown` | | unparseable frame instead of escaping unmapped | `List() was not equal to List("WorkflowErrorEvent")` | | unparseable frame recorded in the metadata store | `List() was not equal to List(COMPILATION_ERROR)` | None of the 15 pre-existing tests regressed. ### Spec comments were updated, not just tests added The spec's header paragraph on malformed frames, its "deliberately not covered" entry for `ModifyLogicRequest`, and two in-test notes all documented these as known-and-unpinned. Leaving them would have left the spec asserting one thing and explaining the opposite, so they are rewritten to match. Trap avoidance, all previously encountered in this file: no assertions on `ClusterListener.numWorkerNodesInCluster` (its default is `0`, so such an assertion passes even against a hard-coded literal); `PrivilegeEnum.WRITE` is fed rather than the `NONE` default; and the new tests use `TestWorkflowService`, which overrides `disconnect()`, so `afterEach` never reaches the null `AmberRuntime._actorSystem` — the same pattern the existing tests use. ### Verification - `WorkflowWebsocketResourceSpec`: **19/19**. - Blast radius: `TexeraWebSocketRequestSpec`, `SessionStateSpec`, `ServletAwareConfiguratorSpec`, `WebsocketInputSpec` — 28/28 across 4 suites. `TexeraWebSocketRequestSpec` pins `InvalidTypeIdException` at the mapper level, which these changes leave untouched. - `scalafmtCheck`, `Test/scalafmtCheck`, `scalafixAll --check` all pass. - Production diff is 15 lines in one file. ### Any related issues, documentation, discussions? Closes #7801 Closes #7454 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.web.resource.WorkflowWebsocketResourceSpec" ``` ``` [info] Suites: completed 1, aborted 0 [info] Tests: succeeded 19, failed 0, canceled 0, ignored 0, pending 0 ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) Co-authored-by: Xuan Gu <[email protected]> --- .../web/resource/WorkflowWebsocketResource.scala | 15 ++- .../resource/WorkflowWebsocketResourceSpec.scala | 125 +++++++++++++++++---- 2 files changed, 114 insertions(+), 26 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/WorkflowWebsocketResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/WorkflowWebsocketResource.scala index 3a02a9a4c2..08df5df322 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/WorkflowWebsocketResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/WorkflowWebsocketResource.scala @@ -78,7 +78,6 @@ class WorkflowWebsocketResource extends LazyLogging { @OnMessage def myOnMsg(session: Session, message: String): Unit = { - val request = objectMapper.readValue(message, classOf[TexeraWebSocketRequest]) val userOpt = session.getUserProperties.asScala .get(classOf[User].getName) .map(_.asInstanceOf[User]) @@ -88,6 +87,9 @@ class WorkflowWebsocketResource extends LazyLogging { val workflowStateOpt = sessionState.getCurrentWorkflowState val executionStateOpt = workflowStateOpt.flatMap(x => Option(x.executionService.getValue)) try { + // Inside the try on purpose: a frame the mapper cannot bind is an error like any other, and + // has to travel back to the client through the same reporting path as a handler failure. + val request = objectMapper.readValue(message, classOf[TexeraWebSocketRequest]) request match { case heartbeat: HeartBeatRequest => sessionState.send(HeartBeatResponse()) @@ -97,7 +99,12 @@ class WorkflowWebsocketResource extends LazyLogging { ) case modifyLogicRequest: ModifyLogicRequest => if (workflowStateOpt.isDefined) { - val executionService = workflowStateOpt.get.executionService.getValue + // Reconfiguration needs the execution, not merely the workflow. `executionService` is a + // BehaviorSubject with no initial value, so reading it through a workflow-shaped guard + // yields null and NPEs one dereference later. + val executionService = executionStateOpt.getOrElse( + throw new IllegalStateException("workflow execution is not initialized") + ) val modifyLogicResponse = executionService.executionReconfigurationService.modifyOperatorLogic( modifyLogicRequest @@ -121,7 +128,9 @@ class WorkflowWebsocketResource extends LazyLogging { case None => throw new IllegalStateException("workflow is not initialized") } case other => - workflowStateOpt.map(_.executionService.getValue) match { + // `executionStateOpt`, not `workflowStateOpt.map(_.executionService.getValue)`: the latter + // wraps a null execution into Some(null), which walks past the None arm below and NPEs. + executionStateOpt match { case Some(value) => value.wsInput.onNext(other, uidOpt) case None => throw new IllegalStateException("workflow execution is not initialized") } diff --git a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala index b059c9ea43..4761c5b302 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala @@ -19,6 +19,7 @@ package org.apache.texera.web.resource +import com.fasterxml.jackson.databind.exc.InvalidTypeIdException import com.google.protobuf.timestamp.Timestamp import org.apache.texera.amber.clustering.ClusterListener import org.apache.texera.amber.core.tuple.{Attribute, AttributeType} @@ -80,11 +81,17 @@ import scala.jdk.CollectionConverters.{IteratorHasAsScala, MapHasAsJava, SeqHasA * - the catch-all error mapper, which turns any exception into a `WorkflowFatalError`, routes it * either to the execution's metadata store or (with no execution) straight to the socket, and * then rethrows so the container sees it too. Both halves are asserted: dropping either the - * send or the rethrow would leave the other silently missing. - * - * Worth knowing, and the reason there is no malformed-frame test here: `objectMapper.readValue` - * sits OUTSIDE the try, so an unparseable frame escapes un-mapped and the client is told nothing. - * That looks like an oversight, but pinning today's behaviour would cement it. + * send or the rethrow would leave the other silently missing. `objectMapper.readValue` runs + * INSIDE the guarded region, so an unparseable frame is reported through that same mapper + * rather than escaping un-mapped with the client told nothing; both of the mapper's routing + * arms are asserted for a parse failure as well as for a handler failure. + * - the three-state distinction every execution-scoped arm has to make: no workflow, a workflow + * with no execution yet, and a live execution. `executionService` is a `BehaviorSubject` + * created without an initial value, so `getValue` is null until one is published, and reading + * it through `workflowStateOpt.map(...)` yields `Some(null)` — a shape that slips past a `case + * None` guard and NPEs on the next dereference. The middle state is covered for both + * `ModifyLogicRequest` and the `case other` fall-through; it is also the only input that tells + * `workflowStateOpt` apart from `executionStateOpt`, so it is what pins each guard's condition. * * Everything here drives a mocked `javax.websocket.Session` (the pattern `CollaborationResourceSpec` * established) and registers a `SessionState` directly, so no real workflow is created. @@ -93,14 +100,10 @@ import scala.jdk.CollectionConverters.{IteratorHasAsScala, MapHasAsJava, SeqHasA * - `myOnOpen`'s missing-`wid`/`cuid` and bogus-privilege paths, which fail with NPE / * IndexOutOfBounds / IllegalArgumentException. A harmless improvement (a default, or a clear * message) would break such a test. - * - `ModifyLogicRequest` with a workflow but no execution: `executionService.getValue` is null, - * so the handler NPEs instead of reporting "workflow execution is not initialized". Same guard - * gap as `case other` below — reported, not pinned. It is also the only input that would tell - * `if (workflowStateOpt.isDefined)` apart from `if (executionStateOpt.isDefined)`, so that - * guard's exact condition is left unpinned on purpose; see the no-workflow case below. * - `ResultPaginationRequest`'s real payload, which needs a DB; only the request/response * pass-through is asserted here, against a stubbed result service. - * - the deserialization line, already owned by `TexeraWebSocketRequestSpec`. + * - what the mapper itself accepts and rejects, already owned by `TexeraWebSocketRequestSpec`; + * the malformed-frame cases here are about the endpoint's handling, not the binding rules. */ class WorkflowWebsocketResourceSpec extends AnyFlatSpec @@ -453,13 +456,8 @@ class WorkflowWebsocketResourceSpec it should "report a runtime command that arrives before any workflow is attached" in { // Anything that is not one of the four named requests falls through to `wsInput`, which only - // exists once an execution has been created. - // - // Note this drives the case with NO workflow attached. With a workflow attached but no - // execution, `executionService.getValue` returns null and `workflowStateOpt.map(...)` yields - // Some(null), which slips past the `case None` guard and NPEs on `value.wsInput`. That is a - // real gap in the guard, but it is an accidental failure mode rather than a contract, so it is - // reported rather than pinned here. + // exists once an execution has been created. This is the no-workflow state; the workflow-but-no- + // execution state is the next case, and both have to report the same thing. val (session, sent) = mockSession(uid = Some(42)) registerState(session, PrivilegeEnum.WRITE) @@ -497,6 +495,27 @@ class WorkflowWebsocketResourceSpec sent shouldBe empty } + it should "report a runtime command that arrives with a workflow but no execution" in { + // The third state of this arm, between the two above: a workflow IS attached, but nothing has + // published an execution into it yet. `executionService` is a `BehaviorSubject` created without + // an initial value, so `getValue` is null, and this is the input that pins how the arm reads it: + // `workflowStateOpt.map(_.executionService.getValue)` would wrap that null into `Some(null)` and + // walk straight past the `case None` guard that exists to report this very situation, NPEing on + // `value.wsInput` instead. The message has to be the same one the no-workflow case above + // produces: the client cannot act on the difference, and an NPE tells it nothing at all. + val (session, sent) = mockSession(uid = Some(42)) + val workflow = new TestWorkflowService(9107L) + attach(session, PrivilegeEnum.WRITE, workflow) + + val ex = intercept[IllegalStateException] { + resource.myOnMsg(session, """{"type":"WorkflowPauseRequest"}""") + } + ex.getMessage should include("workflow execution is not initialized") + + sentTypes(sent) shouldBe Seq("WorkflowErrorEvent") + fatalErrorMessages(sent).head should include("workflow execution is not initialized") + } + // -- reconfiguration ---------------------------------------------------------- it should "forward a ModifyLogicRequest to the execution's reconfiguration service" in { @@ -529,11 +548,9 @@ class WorkflowWebsocketResourceSpec // throw NoSuchElementException, the catch-all would turn it into a WorkflowErrorEvent, and the // rethrow would escape — so both assertions here have teeth. // - // NOT pinned, and it cannot be: rewriting the guard as `if (executionStateOpt.isDefined)` is - // indistinguishable from the current one to every test in this suite. The only input that - // separates them is a workflow with no execution, where today's guard NPEs on - // `executionService.getValue` and the rewrite quietly does nothing. Pinning either answer would - // cement one of them, and the rewrite is arguably the fix. + // Silence is the assertion, and it is what separates this from the next case: with no workflow + // there is nothing to reconfigure and nothing to say, whereas a workflow whose execution has not + // started yet is a state the client can act on and is reported. val (session, sent) = mockSession(uid = Some(42)) registerState(session, PrivilegeEnum.WRITE) @@ -542,6 +559,25 @@ class WorkflowWebsocketResourceSpec sent shouldBe empty } + it should "report a ModifyLogicRequest that arrives before any execution exists" in { + // The input that separates `workflowStateOpt.isDefined` from the execution actually being there: + // a workflow with nothing published into its `executionService`. Reconfiguration needs the + // execution, not the workflow, so the guard has to be about the execution — reading + // `executionService.getValue` behind a workflow-shaped guard yields null and NPEs one dereference + // later. Same message as the `case other` arm, for the same reason. + val (session, sent) = mockSession(uid = Some(42)) + val workflow = new TestWorkflowService(9108L) + attach(session, PrivilegeEnum.WRITE, workflow) + + val ex = intercept[IllegalStateException] { + resource.myOnMsg(session, modifyLogicFrame) + } + ex.getMessage should include("workflow execution is not initialized") + + sentTypes(sent) shouldBe Seq("WorkflowErrorEvent") + fatalErrorMessages(sent).head should include("workflow execution is not initialized") + } + // -- pagination --------------------------------------------------------------- it should "answer a pagination request with what the result service returns" in { @@ -585,6 +621,49 @@ class WorkflowWebsocketResourceSpec sent shouldBe empty } + // -- the error mapper and unparseable frames ---------------------------------- + + it should "report an unparseable frame instead of letting it escape unmapped" in { + // A stale or typo'd client sends an id nothing binds to. The mapper's own rejection is already + // pinned by `TexeraWebSocketRequestSpec`; what this asserts is that the failure travels through + // the endpoint's error mapper on its way out, exactly like a failure raised by a handler. The + // rethrow AND the send are both asserted, for the same reason as the write-access gate: losing + // either one leaves the other silently missing. + val (session, sent) = mockSession(uid = Some(42)) + registerState(session, PrivilegeEnum.WRITE) + + val ex = intercept[InvalidTypeIdException] { + resource.myOnMsg(session, """{"type":"NoSuchWebSocketRequest"}""") + } + ex.getMessage should include("NoSuchWebSocketRequest") + + // No execution is attached, so the mapper's socket arm is the one that runs. + sentTypes(sent) shouldBe Seq("WorkflowErrorEvent") + // The offending id has to reach the client: a generic "bad frame" would leave a stale frontend + // with nothing to debug against. + fatalErrorMessages(sent).head should include("NoSuchWebSocketRequest") + } + + it should "record an unparseable frame in the execution's metadata store when one exists" in { + // The other half of the mapper, reached by a parse failure rather than a handler failure: with an + // execution attached the error is written to its metadata store and nothing goes to the socket. + // Without this case, a fix that reported parse failures by sending directly from the parse site + // instead of routing them through the mapper would look correct. + val (session, sent) = mockSession(uid = Some(42)) + val workflow = new TestWorkflowService(9109L) + val execution = newExecution() + attach(session, PrivilegeEnum.WRITE, workflow, Some(execution)) + + intercept[InvalidTypeIdException] { + resource.myOnMsg(session, """{"type":"NoSuchWebSocketRequest"}""") + } + + val errors = execution.executionStateStore.metadataStore.getState.fatalErrors + errors.map(_.`type`) shouldBe Seq(COMPILATION_ERROR) + errors.head.message should include("NoSuchWebSocketRequest") + sent shouldBe empty + } + // -- the error mapper's metadata-store arm ------------------------------------ it should "record a failure in the execution's metadata store, replacing the stale compilation error" in {
