eugenegujing opened a new pull request, #7609:
URL: https://github.com/apache/texera/pull/7609
### What changes were proposed in this PR?
This PR adds `StartChannelHandlerSpec`, the first unit coverage for
`amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala`.
The handler was last changed by #6913, which adjusted the boundary-state
emission but added only integration coverage (`LoopIntegrationSpec`) and
Python-side unit tests, so the Scala handler's state-emission and exception
paths were unasserted at the unit level.
No production code is changed; this is a test-only PR.
The spec drives a real `DataProcessor` and asserts on the worker's outgoing
messages, so the marker and the emitted state are checked as the wire payloads
a downstream worker would actually receive rather than as mocked calls. The 15
tests pin the handler's three steps and the order between them:
- the input port is resolved from the channel the ECM arrived on, and that
port is what `produceStateOnStart` receives;
- the unaligned START_CHANNEL marker reaches every data channel and no
control channel, after the pending output is flushed;
- the operator's boundary state is emitted to every data channel with the
"no loop" envelope, an empty-but-present state is still emitted, and no state
is emitted when the operator produces none;
- every marker precedes every state, and the marker still goes out first
when the operator throws;
- a sink worker with no data channels emits nothing at all yet still replies
successfully, and a second invocation repeats the whole sequence because the
handler has no once-only guard;
- an operator exception, an operator `Error`, and a failure raised inside
`emitState` are all swallowed, reported through `handleExecutorException`, and
answered with a successful reply, while a `ControlThrowable` escapes and an
unassigned port fails the RPC outright.
Four of these record current behavior that a reader may find surprising, and
each says so in a comment rather than implying endorsement. `ErrorUtils.safely`
swallows `java.lang.Error` because its `OutOfMemoryError` guard is a
commented-out line, the port resolution sits before the `try` so an unassigned
port escapes as an RPC failure instead of being reported like every other
failure one line later, and the RPC replies successfully even after the
operator fails because the failure is surfaced out of band. If any of these is
later changed deliberately, the corresponding test turns red and forces that
decision to be explicit, which is the point of pinning them.
### Any related issues, documentation, discussions?
Closes #7606
### How was this PR tested?
The 15 new tests were run locally, together with the three pre-existing
specs in the same package and the wider worker package:
```
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.worker.promisehandlers.StartChannelHandlerSpec"
-> Tests: succeeded 15, failed 0
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.worker.promisehandlers.*"
-> Suites: completed 4, Tests: succeeded 31, failed 0
sbt "WorkflowExecutionService/testOnly
org.apache.texera.amber.engine.architecture.worker.*"
-> Suites: completed 14, Tests: succeeded 112, failed 0
sbt "WorkflowExecutionService/Test/scalafmtCheck"
-> success
```
Both positive and negative directions are covered, along with the empty and
boundary cases: a produced state and no produced state, an empty-but-present
state, zero downstream data channels, a never-registered input channel, and a
repeated invocation.
The assertions were mutation-checked rather than assumed to be meaningful.
Fifteen mutations were applied to the production code one at a time and every
one of them turned the spec red, including flipping `NO_ALIGNMENT` to
`PORT_ALIGNMENT`, swapping `METHOD_START_CHANNEL` for `METHOD_END_CHANNEL`,
inverting `isDefined`, deleting the `emitState` call, passing a constant port
to `produceStateOnStart`, sending the marker after the `try` block instead of
before it, dropping the `handleExecutorException` call, restricting the marker
to the first data channel only, deleting the `outputManager.flush()` that
precedes the marker, hoisting `emitState` out of the `try`, adding an
idempotence guard, and pausing with a different `PauseType`. 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]
