test(amber): add ClusterListenerSpec against a single-node cluster - #7323
Open
aglinxinyuan wants to merge 3 commits into
Open
test(amber): add ClusterListenerSpec against a single-node cluster#7323aglinxinyuan wants to merge 3 commits into
aglinxinyuan wants to merge 3 commits into
Conversation
ClusterListener was the last file in the engine sitting at 0%. It is worth covering because the count it maintains is what the frontend's cluster badge renders: updateClusterStatus recomputes numWorkerNodesInCluster on every membership event and pushes a ClusterStatusUpdateEvent to every open session, so a listener that stops subscribing or stops fanning out leaves every client showing a stale node count with nothing failing. The suite runs a real single-node cluster. AmberRuntime.pekkoConfig selects the cluster provider with artery on port 0, so joining the node to itself produces genuine MemberUp events. That is not incidental: Member is private[cluster] and cannot be synthesized, so a real join is the only way to reach the event path at all. Three tests: the member-address reply, the recompute-and-fan-out on a membership event, and the catch-all arm. The catch-all test goes through TestActorRef.receive rather than `!`. This matters -- an earlier draft using `!` stayed green with the catch-all replaced by a `throw`, because supervision restarts the actor and a restarted listener answers the next request exactly like one that never failed. TestActorRef.receive lets the exception reach the caller, so "did not throw" means what it says. The subscribe path is likewise mutation-checked: dropping cluster.subscribe from preStart turns the fan-out test red. Two ordering hazards are handled explicitly, both found by failures rather than reasoning. Listeners are stopped at the end of each case, because one left running keeps iterating SessionState.getAllSessionStates and throws ConcurrentModificationException as soon as a later case registers a session. And the mock session is removed inside the case that created it, because ScalaMock scopes expectations per test while the SessionState registry is JVM-global -- a leftover session gets called by a later listener against an expired mock. Left uncovered deliberately: the MemberRemoved recovery arm, which walks every live execution and calls notifyNodeFailure or forcefullyStop, so it needs real Amber clients. Verified across three consecutive runs. No production file is touched.
Contributor
Automated Reviewer SuggestionsBased on the
|
Contributor
There was a problem hiding this comment.
Pull request overview
This PR adds a new ScalaTest spec for Amber’s ClusterListener, exercising its real Pekko Cluster membership event handling in a single-node (self-joined) cluster to cover previously-uncovered logic that drives the frontend’s cluster badge worker count.
Changes:
- Introduces
ClusterListenerSpecthat boots a real single-node Pekko Cluster and validatesGetAvailableNodeAddressesbehavior. - Verifies
MemberUp/initial-state replay triggersupdateClusterStatusto recomputenumWorkerNodesInClusterand fan outClusterStatusUpdateEventto registered websocket sessions. - Adds a catch-all test using
TestActorRef.receiveto ensure unrecognized messages don’t crash the actor.
Suppressed comments (1)
amber/src/test/scala/org/apache/texera/amber/clustering/ClusterListenerSpec.scala:182
sent.toSeqis read while another thread may be appending tosent(seemockSession), which can cause races or inconsistent snapshots. Take a synchronized snapshot before parsing/filtering.
val counts = sent.toSeq
.map(objectMapper.readTree)
.filter(_.get("type").asText() == "ClusterStatusUpdateEvent")
.map(_.get("numWorkers").asInt())
// Every open session is told, and told the recomputed number.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+166
to
+190
| withSession { sent => | ||
| ClusterListener.numWorkerNodesInCluster = -1 | ||
|
|
||
| // Creating the listener subscribes it with InitialStateAsEvents, so the already-Up member is | ||
| // replayed to it as a MemberUp. That is what drives updateClusterStatus here - no synthetic | ||
| // Member is needed, and none could be built (Member is private[cluster]). | ||
| withListener { _ => | ||
| awaitAssert( | ||
| { | ||
| // The count is recomputed from live membership, not left at the sentinel. | ||
| ClusterListener.numWorkerNodesInCluster shouldBe 1 | ||
|
|
||
| val counts = sent.toSeq | ||
| .map(objectMapper.readTree) | ||
| .filter(_.get("type").asText() == "ClusterStatusUpdateEvent") | ||
| .map(_.get("numWorkers").asInt()) | ||
| // Every open session is told, and told the recomputed number. | ||
| counts should not be empty | ||
| counts.last shouldBe 1 | ||
| }, | ||
| 15.seconds, | ||
| 500.millis | ||
| ) | ||
| } | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Agreed — fixed in da9b77b. The previous value is captured and restored in a local finally around that one case, so afterAll is now just a backstop rather than the only restore.
You're right that it matters here specifically: amber sets no Test / fork and suites share one JVM, so the sentinel really was observable to a sibling for the duration of the case.
Comment on lines
+201
to
+204
| val listener = TestActorRef[ClusterListener](Props[ClusterListener]()) | ||
|
|
||
| noException should be thrownBy listener.receive("not a cluster event") | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Good catch, and fixed in da9b77b — the TestActorRef is now stopped in a finally with watch / expectTerminated, the same as the other two cases.
You identified the inconsistency precisely: the other two cases go through a withListener helper that exists for exactly this reason, and this one bypassed it because it needs the TestActorRef for receive rather than !. So it subscribed in preStart like any other listener and then stayed alive — reintroducing the leftover-subscriber hazard the helper was written to avoid.
Re-verified across three consecutive runs after the change, since the ordering hazards in this suite are the kind that only show up on a second pass.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Xinyuan Lin <xinyual3@uci.edu>
…s safe Addresses review feedback, all three points of which were correct: - The collected frames were a plain ArrayBuffer written from the listener's actor thread and read from the test thread inside awaitAssert. It is now a ConcurrentLinkedQueue, and the body receives a snapshot function rather than the live collection, so each assertion reads a consistent point-in-time view. Worth fixing on its own merits: this is the same unsynchronised cross-thread hand-off the spec exists to document on the production side. - The numWorkerNodesInCluster sentinel is now restored in a local finally rather than only in afterAll, so a sibling suite in the shared JVM cannot observe -1 for longer than the one case needs it. - The TestActorRef listener in the catch-all case is now stopped like the others. It subscribes in preStart just the same, and a listener left running keeps iterating the shared SessionState registry on every membership event -- exactly the hazard the rest of the suite avoids. Re-verified across three consecutive runs. No production file is touched.
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 371 | 0.227 | 25,901/40,412/40,412 us | 🔴 +14.9% / 🔴 +160.1% |
| ⚪ | bs=100 sw=10 sl=64 | 782 | 0.477 | 126,650/150,325/150,325 us | ⚪ within ±5% / 🔴 +37.8% |
| ⚪ | bs=1000 sw=10 sl=64 | 887 | 0.541 | 1,130,707/1,165,320/1,165,320 us | ⚪ within ±5% / 🔴 +12.8% |
Baseline details
Latest main 86865f3 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 371 tuples/sec | 388 tuples/sec | 767.32 tuples/sec | -4.4% | -51.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.227 MB/s | 0.237 MB/s | 0.468 MB/s | -4.2% | -51.5% |
| bs=10 sw=10 sl=64 | p50 | 25,901 us | 24,562 us | 12,772 us | +5.5% | +102.8% |
| bs=10 sw=10 sl=64 | p95 | 40,412 us | 35,178 us | 15,538 us | +14.9% | +160.1% |
| bs=10 sw=10 sl=64 | p99 | 40,412 us | 35,178 us | 18,948 us | +14.9% | +113.3% |
| bs=100 sw=10 sl=64 | throughput | 782 tuples/sec | 793 tuples/sec | 972.51 tuples/sec | -1.4% | -19.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.477 MB/s | 0.484 MB/s | 0.594 MB/s | -1.4% | -19.6% |
| bs=100 sw=10 sl=64 | p50 | 126,650 us | 125,575 us | 103,020 us | +0.9% | +22.9% |
| bs=100 sw=10 sl=64 | p95 | 150,325 us | 152,474 us | 109,070 us | -1.4% | +37.8% |
| bs=100 sw=10 sl=64 | p99 | 150,325 us | 152,474 us | 118,964 us | -1.4% | +26.4% |
| bs=1000 sw=10 sl=64 | throughput | 887 tuples/sec | 891 tuples/sec | 1,005 tuples/sec | -0.4% | -11.7% |
| bs=1000 sw=10 sl=64 | MB/s | 0.541 MB/s | 0.544 MB/s | 0.613 MB/s | -0.6% | -11.8% |
| bs=1000 sw=10 sl=64 | p50 | 1,130,707 us | 1,122,336 us | 1,002,400 us | +0.7% | +12.8% |
| bs=1000 sw=10 sl=64 | p95 | 1,165,320 us | 1,167,270 us | 1,039,228 us | -0.2% | +12.1% |
| bs=1000 sw=10 sl=64 | p99 | 1,165,320 us | 1,167,270 us | 1,069,081 us | -0.2% | +9.0% |
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,538.71,200,128000,371,0.227,25900.90,40412.32,40412.32
1,100,10,64,20,2558.12,2000,1280000,782,0.477,126650.17,150324.65,150324.65
2,1000,10,64,20,22556.62,20000,12800000,887,0.541,1130707.35,1165319.88,1165319.88
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #7323 +/- ##
============================================
+ Coverage 83.37% 83.40% +0.03%
- Complexity 4129 4135 +6
============================================
Files 1166 1166
Lines 46424 46424
Branches 5174 5174
============================================
+ Hits 38707 38722 +15
+ Misses 6000 5976 -24
- Partials 1717 1726 +9
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
ClusterListenerwas the last file in the engine at 0%. It is worth covering rather than skipping, because the count it maintains is what the frontend's cluster badge renders:updateClusterStatusrecomputesnumWorkerNodesInClusteron every membership event and pushes aClusterStatusUpdateEventto every open session. A listener that stops subscribing, or stops fanning out, leaves every client showing a stale node count with nothing failing.The suite runs a real single-node cluster.
AmberRuntime.pekkoConfigselects the cluster provider with artery on port 0, so joining the node to itself makes it the leader and produces genuineMemberUpevents. That is not a convenience:Memberisprivate[cluster]and cannot be synthesized, so a real join is the only way to reach the event path at all.Three tests — the member-address reply, the recompute-and-fan-out on a membership event, and the catch-all arm.
The catch-all test goes through
TestActorRef.receive, not!. This one is worth explaining, because the obvious version does not work:The subscribe path is mutation-checked the same way: dropping
cluster.subscribefrompreStartturns the fan-out test red.Two ordering hazards are handled explicitly, both found by failures rather than by reasoning:
SessionState.getAllSessionStateson every membership event — an earlier draft died withConcurrentModificationExceptionas soon as a later case registered a session.SessionStateregistry is JVM-global, so a leftover session gets called by a later listener against an expired mock (Unexpected call: Session.getAsyncRemote).Both of those stem from a real production race, which this PR does not attempt to fix.
SessionState's registry is a plain unsynchronizedmutable.HashMap:and
updateClusterStatusiterates it from the cluster-event thread while websocket open/close mutate it from container threads. A node joining or leaving while a user opens a tab can throw inside the listener. Happy to open a separate issue for it.Left uncovered deliberately: the
MemberRemovedrecovery arm, which walksWorkflowService.getAllWorkflowServicesand callsnotifyNodeFailureorforcefullyStopon each live execution — that needs real Amber clients, i.e. integration scope.No production file is touched.
Any related issues, documentation, discussions?
Closes #7321
How was this PR tested?
Run three times consecutively in one invocation, because the first draft was order-dependent and I wanted the fix demonstrated rather than assumed:
Test/scalafmtCheckandTest/scalafix --checkboth[success].Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)