aglinxinyuan opened a new pull request, #8406:
URL: https://github.com/apache/texera/pull/8406
### What changes were proposed in this PR?
`FlowControlSpec` had a test that asserted nothing:
```scala
"FlowControl" should "trip the size-cap assertion for a message that exceeds
maxByteAllowed" in {
// ... comment conceding it cannot synthesize an oversized payload ...
val fc = new FlowControl()
(1L to 1000L).foreach(i => fc.getMessagesToSend(msg(i)))
succeed
}
```
It sent 1000 messages and then checked no property of the result. The name
claimed a guarantee the file did not pin.
**Measured proof the name was empty.** Deleting the guard the test is named
after — the `assert(creditNeeded <= maxByteAllowed, ...)` block at the top of
`FlowControl.getMessagesToSend` — changed nothing:
| production `FlowControl.scala` | `FlowControlSpec` |
|---|---|
| unmodified | 14 passed, 0 failed |
| `assert(creditNeeded <= maxByteAllowed, ...)` block deleted | 14 passed, 0
failed |
The fixture could never reach the guard: `FixedSizePayload` reports 200
bytes and `flow-control.max-credit-allowed-in-bytes-per-channel` is
1,600,000,000, so `200 <= 1600000000` held on all 1000 iterations. 1000 x 200 =
200,000 bytes does not exhaust credit either, so the messages never took the
stashing path.
**The change: two tests, both with real assertions.**
1. `"FlowControl.getMessagesToSend" should "reject a message larger than the
whole credit cap"` — this one actually trips the guard. An oversized payload
turns out to be cheap to build: `DataFrame.inMemSize` is
`frame.map(_.inMemSize).sum`, which does **not** deduplicate, so an
`Array[Tuple]` holding the same tuple reference N times reports N x its size.
One tuple with a 100,000-char string, repeated `maxBytes / tupleSize + 1`
times, reports over the cap for a few hundred KB of real memory. The test
intercepts the `AssertionError`, checks its message, and checks the rejection
left the channel untouched (`getCredit` unchanged, not marked overloaded —
which is what the out-of-credit branch below the guard would have done instead).
2. `it should "forward every under-cap message and charge its size against
the credit"` — replaces the old body, pinning the fast path the guard sits on:
```
per message i in 1..1000:
getMessagesToSend(msg(i)).toList == List(msg(i)) -- forwarded, not
stashed
getCredit == maxBytes - i * msgSize -- charged exactly its
own size
isOverloaded == false
precondition: batch * msgSize < maxBytes -- else these
expectations
describe the
stashing path
```
Expected values are derived from the fixture (`msgSize` from
`WorkflowMessage.getInMemSize`, `maxBytes` from `ApplicationConfig`), not
hard-coded.
**A claim from my own earlier draft, corrected.** An earlier revision of
this PR stated that covering the size-cap guard "needs an injection seam for
`maxByteAllowed` … that is a production change" and listed the guard as a
disclosed, unavoidable gap. **That was wrong**, and review pressure is what
sent me back to check it. `DataFrame`'s non-deduplicating size sum is the seam,
it is test-only, and the guard is now covered — see the control row in the
mutation table below, which went from PASS to FAIL. No production change was
needed.
**What this PR does NOT do.** It does not change `FlowControl` or any other
`src/main` file (`git diff` on `amber/src/main` is empty). It does not reformat
the file or touch the other 13 tests. Two pre-existing weaknesses in
neighbouring tests are left alone as out of scope: `"eventually drain the stash
across many ack cycles"` ends in `assert(seen == stashed.size)` where `seen` is
incremented once per element of `stashed` in the same loop, so that line is
true by construction; and the suite-constructor `assert(msgSize == 200L)`
hard-codes `WorkflowMessage`'s default, which would abort the whole suite
rather than fail one test if that default ever changed.
### Any related issues, documentation, discussions?
Closes #8402
### How was this PR tested?
All runs: `sbt "WorkflowExecutionService/testOnly ..."` on Java 17, based on
`1cbe857007`.
**Non-vacuity: the new tests can fail, and they catch things the suite did
not already catch.** Four mutations to `FlowControl.getMessagesToSend`, each
run against both the new spec and a verbatim copy of the old `succeed` test
(kept in a throwaway probe suite in the *same* `testOnly` invocation, then
deleted). Failing tests were read from `amber/target/test-reports/TEST-*.xml`
by identity, not from console counts:
| mutation | old `succeed` test | new spec | which identities failed |
|---|---|---|---|
| M1: drop `inflightCredit += creditNeeded` on the fast path | PASS |
**FAIL** (2) | new fast-path test **+ pre-existing** `decreaseInflightCredit
should free credit equal to the acked amount` |
| M2: fast path returns `Iterable.empty` instead of `Iterable(msg)` | PASS |
**FAIL** (2) | new fast-path test **+ pre-existing** `getMessagesToSend should
forward an incoming message when credit is available` |
| M7: `if (inflightCredit == 0) inflightCredit += creditNeeded` — charge
only the first message | PASS | **FAIL** (1) | **new fast-path test only** |
| Control: delete the `assert(creditNeeded <= maxByteAllowed, ...)` block |
PASS | **FAIL** (1) | **new size-cap test only** |
Being explicit about what each row proves, since two of them prove less than
they look:
- M1 and M2 are each **also** caught by one pre-existing neighbour. On those
two rows alone you could not tell whether the rewritten test adds coverage or
merely duplicates it.
- M7 and the control row are the ones that settle it. M7 is a real
behavioural break — flow control stops accounting after the first message —
that **no other test in the file detects**; it fails only because the new test
walks a whole batch instead of one message. The control row is the original
defect: the guard is now pinned, where before nothing in the file noticed its
removal.
Failure messages, for the record:
```
M7 : 1599999800 did not equal 1599999600 after 2 forwarded messages
the credit must be down by 2 * 200
Control : Expected exception java.lang.AssertionError to be thrown, but no
exception was thrown
```
Every mutation was applied from, and reverted to, a pristine copy of
`FlowControl.scala` kept outside the repo (never `git checkout` / `git
restore`), and `git diff -- amber/src/main` was verified empty after each
revert and at the end. The probe suite is deleted; `git status --porcelain`
shows only the one intended test file.
**Regression: baseline first, then compared by failing-test identity, not by
counts.** Scope: every spec in `...architecture.messaginglayer.*` plus
`PekkoMessageTransferServiceSpec`, the only other spec reading the same credit
config, in one invocation.
| run | suites | tests | failed |
|---|---|---|---|
| baseline (file restored to `1cbe857007` content) | 12 | 120 | 0 |
| after this change | 12 | 121 | 0 |
The identity diff is exactly one removal and two additions, with no other
test's name or status changed:
```
- FlowControlSpec :: FlowControl should trip the size-cap assertion for a
message that exceeds maxByteAllowed :: PASS
+ FlowControlSpec :: FlowControl.getMessagesToSend should forward every
under-cap message and charge its size against the credit :: PASS
+ FlowControlSpec :: FlowControl.getMessagesToSend should reject a message
larger than the whole credit cap :: PASS
```
That the multi-run drain test is absent from this diff is the point of one
line in the change: it used `it should`, which bound to the subject of the test
being replaced. It now declares `"FlowControl" should` explicitly, which is why
its identity is byte-identical across the two runs.
Not run: the full `amber` module. Its `@IntegrationTest` suites spawn Python
workers and cannot run on this Windows host, and several Iceberg-backed specs
fail here regardless of the change, so a full-module identity comparison would
have been noise. The change adds no globals or shared state, so the within-JVM
leakage that makes amber's serial execution matter does not apply.
Lint, all clean in the same invocation as the final test run:
`WorkflowExecutionService/scalafmtCheck`,
`WorkflowExecutionService/Test/scalafmtCheck`,
`WorkflowExecutionService/scalafixAll --check`.
### 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]
