aglinxinyuan opened a new pull request, #7118:
URL: https://github.com/apache/texera/pull/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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
