[ISSUE #6461] Follow advertised WebSocket master addresses

[yangjj-iso]

Fixes #6461.

In cluster mode, masterCheck() closes a connection to a non-master Admin but never uses the masterUrl returned in its RUNNING_MODE reply. Retrying only the configured load-balanced URL can repeatedly select the same non-master pod. The check also closes an open connection whose running-mode reply has not arrived yet.

Follow a valid advertised ws/wss master address after an explicit CLUSTER/non-master reply. Deduplicate targets and reuse an existing master connection. Keep the configured seed URLs for later recovery, and construct redirected clients through the existing factory so token, Origin, namespace, gateway port, subscribers, and initial-readiness settings are retained. Invalid addresses fall back to the existing seed retry flow.

Keep open connections while the mode reply is pending. Reset the previous role/address when a socket opens again, make the cross-thread fields visible, and publish the running mode after the reply's address/role have been recorded.

Add 12 regression/guard cases, including real local WebSocket servers that simulate one or two non-master seeds advertising the same master. Verify configuration reaches the subscriber through the master, only one master connection is opened, headers/namespace survive redirection, and seed URLs remain unchanged. Also cover delayed replies, invalid addresses, an already-connected master, seed fallback after a closed redirected connection, and clearing stale role state on reopen. Timers are mocked and checks are invoked directly; no 60-second sleeps are needed.

Validation:

• On the original implementation, 4 of the 12 new cases fail: both real master-redirect scenarios, the delayed-reply case, and the stale-role case.
• After the fix, all 139 selected tests pass (132 in the WebSocket sync module and 7 base handler tests), with zero failures/errors/skips.
• Checkstyle and Apache RAT pass across the affected reactor; git diff --check passes.
• Full-repository install and a deployed Kubernetes/Admin cluster were not run locally.

./mvnw -B -ntp -pl shenyu-sync-data-center/shenyu-sync-data-websocket -am test \
  '-Dtest=WebsocketMasterRedirectTest,WebsocketSyncDataServiceTest,ShenyuWebsocketClientTest,InitialSyncStateTest,InitialSyncConnectionTest,WebsocketConfigTest,*DataHandlerTest' \
  -Dsurefire.failIfNoSpecifiedTests=false -DskipRemoteResources=true \
  -Djacoco.skip=true -Dmaven.javadoc.skip=true
./mvnw -B -ntp -pl shenyu-sync-data-center/shenyu-sync-data-websocket -am apache-rat:check -DskipRemoteResources=true

• Read the contribution guidelines.
• Submit regression tests.
• Full local ./mvnw clean install -Dmaven.javadoc.skip=true (affected-reactor validation above).

The shared E2E WaitForHelperTest uses an ephemeral local HTTP server, including a latch-controlled delayed response, instead of the public Postman Echo service. Both success and timeout tests pass without external HTTP access; production wait/retry behavior is unchanged.

Shared E2E CI regression repairs

The full E2E matrix exposed failures in shared test infrastructure:

• Storage and cluster routing scenarios use a healthchecked local HTTP upstream on the Compose network. The existing URI/method and empty-body assertions exercise the gateway without depending on public httpbin availability.
• RocketMQ uses an explicitly assigned lite-pull consumer with a unique group and request identifier. Wait for topic publication, seek each queue to its beginning, then require the matching access log. Unrelated logs cannot pass the check; non-transient broker errors fail immediately. The proxied request must return HTTP 200.
• Spring Cloud waits for the discovered route to return the order produced by the actual example controller, including its response-name prefix. This covers the delay between selector/rule sync and Eureka's first upstream refresh. Remove an obsolete update using a hardcoded namespace-plugin ID.

Local validation: four HTTP fixture tests (including all six consuming Compose files), seven focused Java tests, Checkstyle and Apache RAT on the affected E2E reactor, and the storage Compose runner checks pass. Regression coverage includes delayed topic publication, rejection of unrelated access logs and broker permission errors, and a route that initially has no usable upstream. Docker daemon execution was unavailable locally; GitHub Actions provides container validation. The four-database storage matrix passed on every branch in the preceding CI run; the latest follow-up reruns the complete matrix.

[Aias00]

Review: Follow advertised WebSocket master addresses (#6461)

Clean, targeted fix for the cluster-mode master-follow problem, with strong test coverage.

What's good

• masterCheck now re-establishes the advertised master connection. Previously a client that was not connected to master was simply closed, so a follower could lose its master link with no recovery. The new LinkedHashSet<URI> masterUrls collects advertised masters during iteration and reconnects them after the iterator completes (avoiding ConcurrentModificationException and duplicate adds).
• Stale-state fix on reconnect. runningMode/masterUrl are now volatile, and onOpen resets runningMode/masterUrl/isConnectedToMaster before sending the RUNNING_MODE request. This fixes a real bug where a reconnected client could carry stale master metadata from a previous connection.
• onMessage ordering fix. The mode is computed into a local mode first; masterUrl/isConnectedToMaster are only set for non-STANDALONE, and runningMode is assigned last — so STANDALONE no longer leaves a stale masterUrl behind.
• Defensive parsing. New parseMasterUri validates ws/wss scheme + non-blank host and logs/ignores invalid addresses instead of blindly reconnecting.
• Tests are thorough: WebsocketMasterRedirectTest exercises real sockets (follower→1 master, multiple followers, pending running-mode reply, invalid address fallback, reuse of existing master connection, and closed-redirect returns to seed), plus testOnOpenClearsPreviousRunningMode for the reset behavior.

Minor / non-blocking

• createClient(masterUri.toString()) relies on the admin URL already carrying the correct path; this matches existing createClient usage so it's fine, just noting the dependency.

CI note

The failing jobs (registry_dependencies, e2e, e2e-storage compose, and the cascade of build (shenyu-integrated-test-*)) are infrastructure/compose failures unrelated to this change — the actual Analyze (java), build, changes, and CodeQL checks pass. Please re-run the infra jobs before merge; they should go green.

Conclusion

Approving. Well-scoped and well-tested fix.

[yangjj-iso]

CI update for head 82398452c: the main ci workflow and the full e2e matrix now pass. The remaining it failure is in build (shenyu-integrated-test-spring-cloud): Maven Central returned 502 Bad Gateway while downloading io.swagger.core.v3:swagger-annotations-jakarta:2.2.22. That job stopped during dependency resolution, before its tests ran.

Could a maintainer with workflow rerun permissions please rerun the failed jobs once dependency downloads are healthy? GitHub rejects reruns from my contributor account with Must have admin rights to Repository. Thank you.

[Aias00]

Review — APPROVE

Two independent changes in one PR, both verified: (1) the #6461 fix to follow advertised WebSocket master addresses, and (2) a set of shared E2E infrastructure repairs. The core fix is small, correct, and backed by tests I confirmed genuinely fail without it. The E2E repairs are empirically validated — every e2e job is green, including the ones that were flaking on other PRs in this queue. No blocking issues.

✅ Core fix — root cause and logic traced

masterCheck() previously closed any client with !isConnectedToMaster(). Since isConnectedToMaster is false from construction and is only set by the RUNNING_MODE reply, a socket whose reply had not yet arrived was closed on every 60s tick — and the advertised masterUrl was never used, so retrying only the configured seeds could keep selecting the same non-master pod.

The new guard is Objects.equals(runningMode, RunningModeEnum.CLUSTER.name()) && !websocketClient.isConnectedToMaster(), where runningMode is the local websocketClient.getRunningMode() (:164). I confirmed:

• The STANDALONE branch immediately above still does timerTask.cancel(); return;, so standalone semantics are unchanged.
• A null runningMode (reply pending) now falls through both branches → the connection is kept. That is precisely the fix.
• parseMasterUri accepts only ws/wss with a non-blank host; anything else is WARNed and returns null, so the client is closed without a redirect and the existing seed retry (which re-seeds once clients empties, i.e. on the next tick) still applies.
• Dedup and reuse work as claimed. LinkedHashSet<URI> gives insertion-ordered dedup, and the second loop's noneMatch(client -> masterUri.equals(client.getURI())) skips a master we are already connected to. I traced both scenarios the PR names: one master + one follower → follower closed, master kept, no new client; two followers advertising the same master → both closed, exactly one new master client.
• createClient is reused for the redirect, so token, Origin, namespace id, gateway port, subscribers and the initial-readiness latch are all retained (confirmed by reading createClient and by the test's header assertions).
• masterCheck is synchronized, and the iterator loop completes before the add loop, so there is no ConcurrentModificationException risk on the backing ArrayList.

✅ Concurrency changes are a correct JMM argument

runningMode/masterUrl become volatile (isConnectedToMaster already was); onOpen resets all three to clear stale role state; and onMessage writes masterUrl and isConnectedToMaster before the volatile runningMode write. Because a volatile write happens-before every subsequent volatile read of the same field that observes it, a reader that sees a non-null runningMode is guaranteed to also see the address and role recorded for that same reply. The PR's "publish the running mode after the reply's address/role have been recorded" is a sound argument, not hand-waving.

✅ The new tests are load-bearing — proven by reverting the fix

Head tree locally: WebsocketMasterRedirectTest 11/11, ShenyuWebsocketClientTest green. I then reverted the fix (original !isConnectedToMaster() branch, non-volatile fields, no onOpen reset, original onMessage field order) and re-ran:

Tests run: 46, Failures: 4, Errors: 0
  WebsocketMasterRedirectTest.testRedirectsFollowerConnectionsToOneMaster[false]  "Configuration should arrive from the advertised master"
  WebsocketMasterRedirectTest.testRedirectsFollowerConnectionsToOneMaster[true]   "Configuration should arrive from the advertised master"
  WebsocketMasterRedirectTest.testKeepsConnectionWhileRunningModeReplyIsPending   expected: <1> but was: <0>
  ShenyuWebsocketClientTest.testOnOpenClearsPreviousRunningMode                   expected: <null> but was: <CLUSTER>

Four failures, and they are exactly the four the PR body names. The remaining eight new cases are guards that correctly hold either way.

Test quality is high: real WebSocketServer instances for the redirect scenarios (no mocks on the path under test), mocked WheelTimerFactory and direct reflective masterCheck() invocation so there are no 60s sleeps, awaitility with explicit aliases, and header/namespace/seed-list assertions that pin the "configuration reaches the subscriber through the master" contract. RocketMQLogConsumptionTest is a pure-mock unit test (no broker) that pins the topic-publication wait, the rejection of unrelated access logs, and the propagation of non-transient broker errors — including the realistic nested-TOPIC_NOT_EXIST shape.

✅ E2E repairs — empirically validated by CI

This is the part I expected to be hand-wavy, and it isn't:

• Every e2e job passed, including e2e-storage on h2/mysql/opengauss/postgres and e2e-case for rocketmq and spring-cloud — i.e. exactly the jobs that were red on #7666 in this same queue.
• WaitForHelperTest (2 tests) and SpringCloudRouteReadinessTest ran green inside the spring-cloud e2e job; RocketMQLogConsumptionTest ran green inside the rocketmq e2e job.
• test_httpbin.py is wired as a CI step in e2e-k8s.yml — which is the workflow that drives the compose matrix, so the placement is right — and it validates both the fixture over HTTP and all six consuming Compose files through docker compose config.
• ${awaitility.version} = 4.0.3, defined at shenyu-e2e/pom.xml:45 and managed at :204, so the new spring-cloud dependency resolves.

I also checked the fixture's surface against its actual consumers rather than trusting the summary: shenyu-e2e-case-storage/DividePluginCases requests /anything, /anything/xx, /anything/xx/yy, /anything/xxx, /anything/200 and /status/200 with GET/POST/PUT/DELETE — all implemented, and test_httpbin.py covers exactly that set. The addNotExists cases are gateway selector/rule mismatches and never reach the upstream, so the fixture's 404 behaviour is not load-bearing for them.

The RocketMQ change fixes the real root cause of the flake I attributed on #7666: a fixed consumer group whose committed offset could already be past this run's log, plus a listener that could be satisfied by a stale message. Now: unique group per run, CONSUME_FROM_FIRST_OFFSET, explicit assign + seekToBegin, a unique e2e=<uuid> marker, and a statusCode(200) assertion on the proxied request. SpringCloudPluginTest drops an update keyed on a hardcoded namespace-plugin ID and replaces it with a readiness wait that asserts the order controller's actual response including its name prefix — a real assertion instead of a sleep.

⚠️ CI attribution — all 10 failures are Maven Central flakes

The jobs that compile and test the changed code all passed:

Job	Result
pr_build (-pl shenyu-sync-data-center/shenyu-sync-data-websocket -am)	pass — 132 tests, 0 failures, 0 Checkstyle violations
e2e aggregate + all e2e-case / e2e-storage matrices	pass
Analyze (java), CodeQL, rat_check, check-license-header	pass

The 10 red jobs are all build (shenyu-integrated-test-*), and the failure is infrastructure, e.g. Could not transfer artifact io.swagger.core.v3:swagger-annotations-jakarta:jar:2.2.22 from/to central: status code: 502 Bad Gateway (502) (plus a 401 from spring-release and "not found" in spring-snapshot). Same class of Maven Central flake as #7663, on a different artifact.

📝 Non-blocking observations

1. masterUri.equals(websocketClient.getURI()) drops the connection without a replacement. If a non-master admin advertises a master URL equal to the client's own URI (contradictory data), the client is closed and removed but the URI is not added to masterUrls, so no replacement is created and that connection is silently lost until a later tick re-seeds — which only happens once clients becomes empty. Adding the URI unconditionally and letting the second loop's noneMatch dedup decide would retry the same address at most once per 60s instead of dropping it. There is no test covering this branch either.

2. The null-runningMode branch removes an old recovery path. A socket whose RUNNING_MODE reply is genuinely lost now stays open and unchecked forever, whereas the old behaviour closed and re-created it every 60s. This is the intended trade-off for the delayed-reply fix and onClose → alreadySync = false → re-send covers the common cases, but it is worth knowing that "admin accepted the socket but never answered RUNNING_MODE" is now a permanent state rather than a self-healing one.

3. String.valueOf(jsonToMap.get(MASTER_URL)) yields the literal "null" when the key is absent, so the WARN reads Ignore invalid websocket master address [null] for what is really an absent field. Pre-existing pattern; Objects.toString(value, null) would read better.

4. parseMasterUri logs the same message twice — once with the exception from URI.create, once without for the scheme/host rejection. Cosmetic.

5. The httpbin fixture implements only GET/POST/PUT/DELETE/HEAD; a future case needing PATCH or OPTIONS would get a 501 from BaseHTTPRequestHandler. Current consumers need only the implemented set. Relatedly, data.decode("utf-8") would raise on a non-UTF-8 body and surface as a dropped connection rather than a clean 400 — not reachable from the current cases.

6. test_compose_uses_healthy_local_upstream hardcodes assertEqual(6, len(files)). Adding a seventh Compose consumer fails this test. Arguably intentional (it forces the fixture test to be extended alongside the wiring), but it is a brittle coupling that deserves a comment saying so.

Verdict

APPROVE. The core fix is minimal, correctly reasoned about concurrency, and its tests provably fail without it; the E2E repairs are backed by a fully green e2e matrix. The notes above are optional polish.