The GitHub Actions job "Required Checks" on texera.git/main has succeeded. Run started by GitHub user github-merge-queue[bot] (triggered by github-merge-queue[bot]).
Head commit for run: 648d54e4e5fae0d34218e8aeb3bdc3cf561955a1 / Xinyuan Lin <[email protected]> test(amber): add specs for the checkpoint and ECM promise handlers (#7118) ### What changes were proposed in this PR? The checkpoint / ECM promise handlers were the engine's lowest-coverage cluster: | Handler | Before | New spec | |---|---|---| | `TakeGlobalCheckpointHandler` | 4.5% | `TakeGlobalCheckpointHandlerSpec` (6 tests) | | `FinalizeCheckpointHandler` | 4.5% | `FinalizeCheckpointHandlerSpec` (5) | | `PrepareCheckpointHandler` | 4.8% | `PrepareCheckpointHandlerSpec` (4) | | `EmbeddedControlMessageHandler` | 8.3% | `EmbeddedControlMessageHandlerSpec` (6) | | `EvaluatePythonExpressionHandler` | 8.3% | `EvaluatePythonExpressionHandlerSpec` (6) | All five follow the existing precedent in this layer (`RetrieveWorkflowStateHandlerSpec`): a real `CoordinatorProcessor` plus a second handler initializer, capturing what the output handler emits — so the tests drive the **real** handler methods, not a re-implementation. Storage uses `ram://` in-memory VFS per the `LoggingSpec` / `DPThreadSpec` precedent, with a distinct folder per case since `VFS.getManager` is a JVM-wide singleton. Covered: the single-physical-op guard; the already-taken short-circuit that downgrades a checkpoint to an estimate; the per-checkpoint subfolder that `WorkflowActor.setupReplay` later resolves; channel scoping and the "latest execution" worker selection; the estimation branches on both worker handlers; the `CheckpointSupport` hand-off and the observable consequence of its skip branch; and the fan-in that must not answer before every targeted worker has replied. **Assertion strength was measured, not asserted.** Four rounds of production mutation, 21 of 22 non-trivial cases shown to go red, every mutation then reverted (`git diff -- amber/src/main` is empty): | Mutation | Result | |---|---| | `physicalOps.size != 1` → `> 2` | 2 tests fail | | `msg.targetOps.flatMap` → `msg.scope.flatMap` | 1 | | `if (!msg.estimationOnly)` → inverted | 3 | | `findLast` → `find` (latest execution) | 1 | | channel-scope filter `&&` → `||` | 1 | | `markAsReplayDestination` hoisted before `collect` | 1 | | drop the already-taken short-circuit | 1 | | `checkpoints.contains` → `!contains` | 5 | | fan-in → fire-and-forget | 1 | | …and 11 more | | Also removes the two commented-out blocks in `CheckpointSpec` (−73 lines), verified dead rather than assumed: the first matches `OpExecInitInfoWithCode` / `OpExecInitInfoWithFunc`, names that a repo-wide grep finds **only inside those comment lines** (`OpExecInitInfo` is now a ScalaPB sealed oneof), and casts to `CheckpointSupport`, which no production operator implements — plus it has zero assertions and one branch is literally `???`. Its intent is already covered by `SerializationManagerSpec` and `CheckpointSubsystemSpec`. The second calls `AmberClient` with six constructor arguments where the signature takes five, and starts a real two-operator workflow with 30s awaits. **Two production bugs surfaced while writing these; neither is fixed here, and neither is pinned as current behaviour.** 1. **`TakeGlobalCheckpointHandler` under-reports `totalSize`.** It accumulates into a captured `var` via a per-future `.onSuccess { resp => totalSize += resp.size }` registered *before* `Future.collect` wraps those futures. A `com.twitter.util.Promise` runs continuations in **reverse** registration order, so for the last-satisfied future the `collect` continuation — and the `.map` that reads the counter — runs before that future's `onSuccess` has contributed. Observed directly: two workers reporting 4000 and 56 produced `TakeGlobalCheckpointResponse(4000)`; a single worker reporting 7 produced `0`. Fix direction is to sum from the collected `Seq` inside the `.map`. The spec case is named for the fan-in it does verify and carries a comment explaining why the total is deliberately not asserted. 2. **A `CheckpointState` can never be written to storage with the shipped config.** `SequentialRecordWriter.writeRecord` does `AmberRuntime.serde.serialize(obj).get`, but `CheckpointState` is a plain `class` that does not extend `java.io.Serializable`, while `cluster.conf` sets `allow-java-serialization = off` with `"java.io.Serializable" = kryo` as the only binding — so the write throws `NotSerializableException`. That kills both persist paths, which means global checkpointing cannot produce a checkpoint on disk and the checkpoint-restore branch of `WorkflowActor.setupReplay` has nothing to read. Existing specs missed it because they only exercise `chkpt.save` / `chkpt.load` (`SerializedState` *is* a case class). Consequence for this PR: the two write-branch cases wrap the call in `Try` and assert only the state the handler mutates beforehand — assertions that hold both before and after a fix, so they are not inverted change-detectors. The class scaladoc says so explicitly. A third, smaller hazard is recorded in the notes but not tested: `EvaluatePythonExpressionHandler` statically types worker results as `EvaluatedValue`, which is not a member of the `ControlReturn` sealed oneof. Erasure hides it on the JVM side and the Python side sets a stray attribute, so an empty `ControlReturn` comes back. That is why the spec asserts dispatch and fan-in shape but never fulfils a worker promise with a fake `EvaluatedValue` — doing so would cement a type confusion. ### Any related issues, documentation, discussions? Closes #7117 ### How was this PR tested? Five new specs, 27 tests. Run together with the three adjacent checkpoint specs they must not duplicate or break — 49 tests, Java 17: ``` sbt "WorkflowExecutionService/testOnly *EvaluatePythonExpressionHandlerSpec *EmbeddedControlMessageHandlerSpec *TakeGlobalCheckpointHandlerSpec *PrepareCheckpointHandlerSpec *FinalizeCheckpointHandlerSpec *CheckpointSpec *SerializationManagerSpec *CheckpointSubsystemSpec" ``` ``` [info] Suites: completed 8, aborted 0 [info] Tests: succeeded 49, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. ``` Isolation was checked too: the new specs were re-run interleaved with `LoggingSpec`, `WorkflowWorkerSpec`, `DPThreadSpec` and the sibling promise-handler specs (16 suites, concurrent in one JVM), and then a second time in the same JVM to confirm the `ram://` folders leave no residue. No new spec mutates `AmberRuntime`'s private `_serde` / `_actorSystem`. `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) Report URL: https://github.com/apache/texera/actions/runs/30657591099 With regards, GitHub Actions via GitBox
