test(amber): make FlowControlSpec's size-cap test actually assert

[aglinxinyuan]

What changes were proposed in this PR?

FlowControlSpec had a test that asserted nothing:

"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)

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Ma77Ball
  You can notify them by mentioning @Ma77Ball in a comment.

[Copilot]

🟢 Approval recommended

The changes are test-only, correctly target previously uncovered behavior with deterministic assertions, and do not introduce fragility or production impact.

Pull request overview

This PR strengthens Amber’s FlowControlSpec by replacing a vacuous “size-cap” test with two concrete, state-asserting unit tests that (1) actually exercise the getMessagesToSend size-cap guard and (2) pin the fast-path credit-accounting behavior across a batch.

Changes:

• Add a new test that constructs an oversized DataFrame payload and asserts FlowControl.getMessagesToSend rejects it via the size-cap assert, without mutating channel credit/overload state.
• Replace the prior assertion-free loop with a batch test that asserts per-message forwarding, exact credit decrement, and non-overloaded state while under the cap.
• Make the “multi-run” stash-drain test bind explicitly to "FlowControl" should ... to keep its test identity stable after inserting the new "FlowControl.getMessagesToSend" should ... block.

File summaries

File	Description
amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/FlowControlSpec.scala	Replaces a non-asserting test with two real assertions covering the size-cap guard and fast-path credit accounting, plus a minor spec-subject adjustment to preserve existing test naming.

Review details

• Files reviewed: 1/1 changed files
• Comments generated: 0
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 7 worse · ⚪ 8 noise (<±5%) · 0 without baseline

Compared against main 1cbe857 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	347	0.212	28,255/39,511/39,511 us	🔴 +23.7% / 🔴 +147.3%
⚪	bs=100 sw=10 sl=64	779	0.475	125,822/151,108/151,108 us	⚪ within ±5% / 🔴 +38.3%
🔴	bs=1000 sw=10 sl=64	894	0.546	1,109,671/1,204,610/1,204,610 us	🔴 +6.2% / 🔴 +15.5%

Baseline details

Latest main 1cbe857 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	347 tuples/sec	425 tuples/sec	755.36 tuples/sec	-18.4%	-54.1%
bs=10 sw=10 sl=64	MB/s	0.212 MB/s	0.26 MB/s	0.461 MB/s	-18.5%	-54.0%
bs=10 sw=10 sl=64	p50	28,255 us	22,838 us	12,938 us	+23.7%	+118.4%
bs=10 sw=10 sl=64	p95	39,511 us	35,118 us	15,980 us	+12.5%	+147.3%
bs=10 sw=10 sl=64	p99	39,511 us	35,118 us	19,233 us	+12.5%	+105.4%
bs=100 sw=10 sl=64	throughput	779 tuples/sec	793 tuples/sec	976.3 tuples/sec	-1.8%	-20.2%
bs=100 sw=10 sl=64	MB/s	0.475 MB/s	0.484 MB/s	0.596 MB/s	-1.9%	-20.3%
bs=100 sw=10 sl=64	p50	125,822 us	124,102 us	102,340 us	+1.4%	+22.9%
bs=100 sw=10 sl=64	p95	151,108 us	152,789 us	109,262 us	-1.1%	+38.3%
bs=100 sw=10 sl=64	p99	151,108 us	152,789 us	118,827 us	-1.1%	+27.2%
bs=1000 sw=10 sl=64	throughput	894 tuples/sec	922 tuples/sec	1,006 tuples/sec	-3.0%	-11.2%
bs=1000 sw=10 sl=64	MB/s	0.546 MB/s	0.563 MB/s	0.614 MB/s	-3.0%	-11.1%
bs=1000 sw=10 sl=64	p50	1,109,671 us	1,086,151 us	999,855 us	+2.2%	+11.0%
bs=1000 sw=10 sl=64	p95	1,204,610 us	1,134,249 us	1,042,833 us	+6.2%	+15.5%
bs=1000 sw=10 sl=64	p99	1,204,610 us	1,134,249 us	1,070,722 us	+6.2%	+12.5%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,575.77,200,128000,347,0.212,28254.95,39511.20,39511.20
1,100,10,64,20,2569.04,2000,1280000,779,0.475,125822.19,151108.19,151108.19
2,1000,10,64,20,22376.52,20000,12800000,894,0.546,1109670.60,1204610.29,1204610.29

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.02%. Comparing base (1cbe857) to head (ec8576d).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #8406      +/-   ##
============================================
- Coverage     94.03%   94.02%   -0.01%     
+ Complexity     4821     4820       -1     
============================================
  Files          1204     1204              
  Lines         48991    48991              
  Branches       5956     5956              
============================================
- Hits          46067    46066       -1     
- Misses         1458     1459       +1     
  Partials       1466     1466              

Flag	Coverage Δ		*Carryforward flag
access-control-service	81.00% <ø> (ø)		Carriedforward from 1cbe857
agent-service	99.32% <ø> (ø)		Carriedforward from 1cbe857
amber	89.91% <ø> (-0.01%)	⬇️
computing-unit-managing-service	73.67% <ø> (ø)		Carriedforward from 1cbe857
config-service	87.12% <ø> (ø)		Carriedforward from 1cbe857
file-service	87.91% <ø> (ø)		Carriedforward from 1cbe857
frontend	96.79% <ø> (ø)		Carriedforward from 1cbe857
notebook-migration-service	83.57% <ø> (ø)		Carriedforward from 1cbe857
pyamber	98.47% <ø> (ø)		Carriedforward from 1cbe857
workflow-compiling-service	77.19% <ø> (ø)		Carriedforward from 1cbe857

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.