eugenegujing opened a new pull request, #7610:
URL: https://github.com/apache/texera/pull/7610
### What changes were proposed in this PR?
This PR adds `WorkerStateUpdatedHandlerSpec`, the first unit coverage for
`amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/WorkerStateUpdatedHandler.scala`.
The handler had no references anywhere in the test tree. The version-ordered
state model it feeds was fixed and tested in #6011, which gave
`WorkerExecution` a spec, but the handler wiring itself was unpinned.
No production code is changed; this is a test-only PR.
The spec drives a real `CoordinatorProcessor` with no ActorSystem and
captures the dispatched client events through the coordinator's output handler.
The 20 tests pin the handler's lookup, the update it applies, and the
broadcasts it emits:
- the physical operator is derived from `ctx.sender`, and the update lands
on that worker alone rather than on the operator's other workers or on another
operator;
- reports are ordered by the request's `stateVersion`, so a stale or
re-delivered version is ignored while a strictly newer one applies, and the
version each worker carries is its own rather than shared across the operator;
- a terminal state absorbs every later report, reached both through a
COMPLETED report and through a TERMINATED one, and the reports refused this way
are still broadcast;
- only running region executions are consulted, and only the first one
owning the operator, while the broadcast statistics span every region execution
including completed ones;
- the skip branch this coverage exists for: when no running region execution
owns the sender's operator the report is dropped, yet `ExecutionStatsUpdate`
and `RuntimeStatisticsPersist` still fire and the reply is still an empty
success;
- both boundaries of the version guard, since `lastStateVersion` starts at
`-1` and the comparison is strictly-greater: a first report at `-1` is dropped
as if stale, and a report at `Long.MaxValue` freezes a worker even when its
state is not terminal.
Two of these record current behavior that a reader may find surprising, and
each says so in a comment rather than implying endorsement. An unknown worker
of a known operator throws a `NullPointerException`, because
`getWorkerExecution` is a `ConcurrentHashMap.get` whose `null` is dereferenced
immediately, and on that path both client broadcasts are lost, which is the
opposite of how an unknown operator is handled one branch away. A report
carrying the proto default `UNINITIALIZED` at version 0 is applied invisibly
and consumes the first real version, so the worker's genuine first transition
is then dropped without a trace. If either is later changed deliberately, the
corresponding test turns red and forces that decision to be explicit.
### Any related issues, documentation, discussions?
Closes #7607
Related: #6011 introduced the version-ordered state model this handler feeds.
### How was this PR tested?
The 20 new tests were run locally, together with the six pre-existing specs
in the same package and the wider coordinator package:
```
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.coordinator.promisehandlers.WorkerStateUpdatedHandlerSpec"
-> Tests: succeeded 20, failed 0
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.coordinator.promisehandlers.*"
-> Suites: completed 7, Tests: succeeded 48, failed 0
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.coordinator.*"
-> Suites: completed 19, Tests: succeeded 171, failed 0
sbt "WorkflowExecutionService/Test/scalafmtCheck"
-> success
```
Both positive and negative directions are covered, along with the empty and
boundary cases: an applied update and a skipped one, a workflow execution with
no region execution at all, the `-1` version sentinel, and the `Long.MaxValue`
ceiling.
The assertions were mutation-checked rather than assumed to be meaningful.
Mutations were applied one at a time to `WorkerStateUpdatedHandler.scala`,
`WorkerExecution.scala` and `OperatorExecution.scala`, and each turned the spec
red: dropping either `sendToClient` call, swapping their order, guarding them
behind a non-empty statistics map, replacing `getRunningRegionExecutions` with
all region executions, widening `find` to `filter`, deriving the operator from
`ctx.receiver` instead of `ctx.sender`, computing the statistics snapshot
before applying the update, forwarding a constant version to `updateState`,
moving the sentinel off `-1`, changing the version comparison to `<=` or `!=`,
adding a state-equality condition to the guard, removing `TERMINATED` from the
terminal check, and returning one shared `WorkerExecution` for every worker.
The production files were restored and verified byte-identical to `HEAD`
afterwards.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude 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]
