This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8406-2a3bba15b2a913d9fe2f6188d029b61dce99d08b in repository https://gitbox.apache.org/repos/asf/texera.git
commit 09b4e74560a0cb16c557816711639493564f9d3c Author: Xinyuan Lin <[email protected]> AuthorDate: Sat Sep 5 06:56:08 2026 +0000 test(amber): make FlowControlSpec's size-cap test actually assert (#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) --- .../messaginglayer/FlowControlSpec.scala | 78 +++++++++++++++++----- 1 file changed, 62 insertions(+), 16 deletions(-) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/FlowControlSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/FlowControlSpec.scala index 418a02426c..d17dd3f22b 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/FlowControlSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/FlowControlSpec.scala @@ -20,9 +20,11 @@ package org.apache.texera.amber.engine.architecture.messaginglayer import org.apache.texera.common.config.ApplicationConfig +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} import org.apache.texera.amber.engine.architecture.common.WorkflowActor.NetworkMessage import org.apache.texera.amber.engine.common.ambermessage.{ + DataFrame, WorkflowFIFOMessage, WorkflowFIFOMessagePayload, WorkflowMessage @@ -49,6 +51,11 @@ class FlowControlSpec extends AnyFlatSpec { private val maxBytes = ApplicationConfig.maxCreditAllowedInBytesPerChannel + // One-field schema, used only to build the oversized DataFrame payload that + // trips the size-cap guard. + private val payloadAttr = new Attribute("payload", AttributeType.STRING) + private val payloadSchema: Schema = Schema().add(payloadAttr) + "FlowControl" should "report full credit and not be overloaded initially" in { val fc = new FlowControl() assert(fc.getCredit == maxBytes) @@ -116,6 +123,33 @@ class FlowControlSpec extends AnyFlatSpec { assert(fc.isOverloaded, "stash still has msg(2L), so overloaded must remain true") } + // The fast path below the size-cap guard: while every message stays under the + // cap, each one is forwarded immediately rather than stashed, is charged exactly + // its own size against the credit, and never flips the overloaded flag. Running + // a whole batch rather than a single message is what catches accounting that is + // right for the first message and then drifts. (The guard itself is covered by + // "reject a message larger than the whole credit cap" further down.) + it should "forward every under-cap message and charge its size against the credit" in { + val batch = 1000L + // Fixture precondition: the whole batch must fit under the cap, otherwise the + // expectations below would be describing the stashing path instead. + assert( + batch * msgSize < maxBytes, + s"fixture no longer exercises the fast path: $batch * $msgSize >= $maxBytes" + ) + + val fc = new FlowControl() + (1L to batch).foreach { i => + val out = fc.getMessagesToSend(msg(i)).toList + assert(out == List(msg(i)), s"message $i must be forwarded, not stashed") + assert( + fc.getCredit == maxBytes - i * msgSize, + s"after $i forwarded messages the credit must be down by $i * $msgSize" + ) + assert(!fc.isOverloaded, s"must not be overloaded after $i under-cap messages") + } + } + "FlowControl.updateQueuedCredit" should "shrink the available credit" in { val fc = new FlowControl() fc.updateQueuedCredit(100L) @@ -146,25 +180,37 @@ class FlowControlSpec extends AnyFlatSpec { // Edge / invalid-input cases — credit math under abnormal conditions // --------------------------------------------------------------------------- - "FlowControl" should "trip the size-cap assertion for a message that exceeds maxByteAllowed" in { - // Build a payload whose getInMemSize returns a value larger than the - // configured per-channel cap. We do this by ratcheting up the Pekko-side - // size accounting via an oversized DataFrame stand-in: emulate by - // exhausting credit to <= 0 and then sending a payload that's already - // larger than 0 — but the assertion in source compares creditNeeded - // against `maxByteAllowed`, not credit. Since FixedSizePayload is 200L - // and maxByteAllowed is multi-GB, we cannot synthesize a too-big payload - // in unit-test scope without producing terabytes. Instead, lock down - // the *guard* shape: a message at exactly maxByteAllowed is allowed by - // the assertion (not strictly greater), so any 200L payload always - // passes — confirm that 1000 sequential 200L messages all pass the - // assertion regardless of credit accounting. + // The size-cap guard at the top of getMessagesToSend. A payload larger than the + // entire per-channel cap could never be sent, so FlowControl rejects it outright + // rather than stashing it forever. Reaching a multi-GB reported size costs + // almost no memory: DataFrame.inMemSize sums Tuple.inMemSize across its array + // without deduplicating, so an array holding the same tuple reference many + // times reports an oversized payload while allocating one tuple and N pointers. + "FlowControl.getMessagesToSend" should "reject a message larger than the whole credit cap" in { + val bigTuple = Tuple.builder(payloadSchema).add(payloadAttr, "x" * 100000).build() + val copies = (maxBytes / bigTuple.inMemSize + 1).toInt + val oversized = NetworkMessage( + 1L, + WorkflowFIFOMessage(channelId, 1L, DataFrame(Array.fill(copies)(bigTuple))) + ) + // Fixture precondition: the payload really is over the cap, so the guard is + // the thing under test rather than the out-of-credit branch below it. + val reportedSize = WorkflowMessage.getInMemSize(oversized.internalMessage) + assert( + reportedSize > maxBytes, + s"fixture payload is not oversized: $reportedSize <= $maxBytes" + ) + val fc = new FlowControl() - (1L to 1000L).foreach(i => fc.getMessagesToSend(msg(i))) - succeed + val thrown = intercept[AssertionError](fc.getMessagesToSend(oversized)) + assert(thrown.getMessage.contains("too big to send through flow control")) + // The rejection must leave the channel untouched — not half-charged, and not + // marked overloaded as the out-of-credit branch would have done. + assert(fc.getCredit == maxBytes) + assert(!fc.isOverloaded) } - it should "eventually drain the stash across many ack cycles (multi-run)" in { + "FlowControl" should "eventually drain the stash across many ack cycles (multi-run)" in { val fc = new FlowControl() // Saturate credit and stash a batch of messages. fc.updateQueuedCredit(maxBytes)
