aglinxinyuan opened a new pull request, #7323: URL: https://github.com/apache/texera/pull/7323
### What changes were proposed in this PR? `ClusterListener` was 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: `updateClusterStatus` recomputes `numWorkerNodesInCluster` on every membership event and pushes a `ClusterStatusUpdateEvent` to 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.pekkoConfig` selects the cluster provider with artery on port 0, so joining the node to itself makes it the leader and produces genuine `MemberUp` events. That is not a convenience: `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`, not `!`.** This one is worth explaining, because the obvious version does not work: > An earlier draft sent the stray message with `!` and asserted the listener still answered. It stayed **green** with the catch-all replaced by a `throw` — supervision restarts the actor, and a restarted listener answers the next request exactly like one that never failed. `TestActorRef.receive` invokes receive directly and lets the exception reach the caller, so "did not throw" means what it says. The subscribe path is mutation-checked the same way: dropping `cluster.subscribe` from `preStart` turns the fan-out test red. **Two ordering hazards are handled explicitly, both found by failures rather than by reasoning:** 1. Listeners are stopped at the end of each case. One left running stays subscribed and keeps iterating `SessionState.getAllSessionStates` on every membership event — an earlier draft died with `ConcurrentModificationException` as soon as a later case registered a session. 2. The mock session is removed inside the case that created it. ScalaMock scopes expectations per test while the `SessionState` registry 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 unsynchronized `mutable.HashMap`: ```scala private val sessionIdToSessionState = new mutable.HashMap[String, SessionState]() def getAllSessionStates: Iterable[SessionState] = sessionIdToSessionState.values ``` and `updateClusterStatus` iterates 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 `MemberRemoved` recovery arm, which walks `WorkflowService.getAllWorkflowServices` and calls `notifyNodeFailure` or `forcefullyStop` on 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? ``` sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.clustering.ClusterListenerSpec" ``` Run **three times consecutively** in one invocation, because the first draft was order-dependent and I wanted the fix demonstrated rather than assumed: ``` [info] Tests: succeeded 3, failed 0, canceled 0, ignored 0, pending 0 [info] Tests: succeeded 3, failed 0, canceled 0, ignored 0, pending 0 [info] Tests: succeeded 3, failed 0, canceled 0, ignored 0, pending 0 ``` `Test/scalafmtCheck` and `Test/scalafix --check` both `[success]`. ### 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]
