This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 40e6879b7d Fixes #7313: Close stalled websocket sessions and bound the
admin send queue (#7343)
40e6879b7d is described below
commit 40e6879b7d3884b7f8874fc09b8775924297d849
Author: BobSong <[email protected]>
AuthorDate: Thu Oct 1 11:02:15 2026 +0800
Fixes #7313: Close stalled websocket sessions and bound the admin send
queue (#7343)
---
.../listener/websocket/WebsocketCollector.java | 144 +++++++++++++++++++--
.../listener/websocket/WebsocketCollectorTest.java | 87 ++++++++++++-
.../logging/rocketmq/DividePluginTest.java | 19 ++-
.../logging/rocketmq/LoggingRuleSyncTest.java | 79 +++++++++++
4 files changed, 319 insertions(+), 10 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
index 968f17bc7c..e100045639 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
@@ -44,6 +44,7 @@ import jakarta.websocket.OnClose;
import jakarta.websocket.OnError;
import jakarta.websocket.OnMessage;
import jakarta.websocket.OnOpen;
+import jakarta.websocket.SendResult;
import jakarta.websocket.Session;
import jakarta.websocket.server.ServerEndpoint;
@@ -56,6 +57,10 @@ import java.util.Queue;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
/**
* The type Websocket data changed listener.
@@ -73,6 +78,25 @@ public class WebsocketCollector {
private static final Map<Session, SessionSendQueue> SESSION_SEND_QUEUES =
Maps.newConcurrentMap();
+ private static final long DEFAULT_SEND_TIMEOUT_MILLIS = 30_000L;
+
+ private static final int DEFAULT_MAX_QUEUED_MESSAGES = 256;
+
+ /**
+ * Watchdog that detects async sends whose container callback never runs,
+ * e.g. on half-open connections, and closes the session so the gateway
+ * reconnects and performs a full synchronization.
+ */
+ private static final ScheduledExecutorService SEND_WATCHDOG =
Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "shenyu-websocket-send-watchdog");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ private static volatile long sendTimeoutMillis =
DEFAULT_SEND_TIMEOUT_MILLIS;
+
+ private static volatile int maxQueuedMessages =
DEFAULT_MAX_QUEUED_MESSAGES;
+
/**
* Namespace captured at registration. {@code Session#isOpen()} is already
false when
* {@code @OnClose} runs, so the namespace cannot be read from the session
at teardown.
@@ -412,6 +436,32 @@ public class WebsocketCollector {
}
}
+ /**
+ * Adjust the per-message send timeout, intended for tests.
+ *
+ * @param timeoutMillis the send timeout in milliseconds
+ */
+ static void setSendTimeoutMillis(final long timeoutMillis) {
+ sendTimeoutMillis = timeoutMillis;
+ }
+
+ /**
+ * Adjust the per-session queued message limit, intended for tests.
+ *
+ * @param limit the maximum number of queued messages
+ */
+ static void setMaxQueuedMessages(final int limit) {
+ maxQueuedMessages = limit;
+ }
+
+ /**
+ * Reset the send timeout and queue limit to their defaults, intended for
tests.
+ */
+ static void resetSendGuards() {
+ sendTimeoutMillis = DEFAULT_SEND_TIMEOUT_MILLIS;
+ maxQueuedMessages = DEFAULT_MAX_QUEUED_MESSAGES;
+ }
+
private static final class InitialSync {
private final Session session;
@@ -432,6 +482,10 @@ public class WebsocketCollector {
private final Queue<String> messages = new ArrayDeque<>();
+ private ScheduledFuture<?> timeoutFuture;
+
+ private String inFlightMessage;
+
private boolean sending;
private boolean closed;
@@ -446,6 +500,14 @@ public class WebsocketCollector {
if (closed) {
return;
}
+ if (messages.size() >= WebsocketCollector.maxQueuedMessages) {
+ // silently dropping messages is not acceptable: close the
session
+ // so the gateway reconnects and performs a full
synchronization
+ LOG.error("websocket send queue overflow on session {},
queued {}, closing session for resync",
+ session.getId(), messages.size());
+ forceClose("send queue overflow");
+ return;
+ }
messages.offer(message);
if (!sending) {
sending = true;
@@ -459,6 +521,7 @@ public class WebsocketCollector {
private void sendNext() {
final String message;
+ final ScheduledFuture<?> future;
synchronized (this) {
if (closed) {
sending = false;
@@ -470,23 +533,88 @@ public class WebsocketCollector {
sending = false;
return;
}
+ inFlightMessage = message;
+ future = SEND_WATCHDOG.schedule(
+ () -> onSendTimeout(message), sendTimeoutMillis,
TimeUnit.MILLISECONDS);
+ timeoutFuture = future;
}
+ boolean submitted;
try {
- session.getAsyncRemote().sendText(message, result -> {
- if (!result.isOK()) {
- LOG.error("websocket send result is exception: ",
result.getException());
- }
- sendNext();
- });
+ session.getAsyncRemote().sendText(message, result ->
onSendResult(future, result));
+ submitted = true;
} catch (RuntimeException ex) {
- LOG.error("websocket send result is exception: ", ex);
- sendNext();
+ LOG.error("websocket send failed synchronously on session {}",
session.getId(), ex);
+ submitted = false;
+ }
+ if (!submitted) {
+ future.cancel(false);
+ forceClose("synchronous send failure");
+ }
+ }
+
+ private void onSendResult(final ScheduledFuture<?> future, final
SendResult result) {
+ future.cancel(false);
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ if (timeoutFuture == future) {
+ timeoutFuture = null;
+ }
+ inFlightMessage = null;
+ if (!result.isOK()) {
+ LOG.error("websocket send result is exception on session
{}, closing session for resync",
+ session.getId(), result.getException());
+ forceClose("send failure");
+ return;
+ }
+ }
+ sendNext();
+ }
+
+ private void onSendTimeout(final String scheduledMessage) {
+ synchronized (this) {
+ if (closed || !sending || !Objects.equals(inFlightMessage,
scheduledMessage)) {
+ return;
+ }
+ LOG.error("websocket send callback not observed within {} ms
on session {},"
+ + " treating the connection as broken and
closing it for resync",
+ sendTimeoutMillis, session.getId());
+ forceClose("send timeout");
+ }
+ }
+
+ private void forceClose(final String reason) {
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ sending = false;
+ inFlightMessage = null;
+ if (Objects.nonNull(timeoutFuture)) {
+ timeoutFuture.cancel(false);
+ timeoutFuture = null;
+ }
+ messages.clear();
+ }
+ LOG.warn("closing websocket session {} to force gateway resync,
reason={}", session.getId(), reason);
+ removeSessionIndexes(session);
+ try {
+ session.close();
+ } catch (Exception ex) {
+ LOG.warn("error closing websocket session {}: {}",
session.getId(), ex.getMessage());
}
}
private void close() {
synchronized (this) {
closed = true;
+ inFlightMessage = null;
+ if (Objects.nonNull(timeoutFuture)) {
+ timeoutFuture.cancel(false);
+ timeoutFuture = null;
+ }
messages.clear();
}
}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
index bf038a24f7..38d5d7fbb4 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
@@ -112,6 +112,7 @@ public final class WebsocketCollectorTest {
websocketCollector = new WebsocketCollector();
// Clear shared static state between tests
clearStaticSessionState();
+ WebsocketCollector.resetSendGuards();
when(session.isOpen()).thenReturn(true);
Map<String, Object> userProperties = new HashMap<>();
userProperties.put(Constants.SHENYU_NAMESPACE_ID,
Constants.SYS_DEFAULT_NAMESPACE_ID);
@@ -549,7 +550,7 @@ public final class WebsocketCollectorTest {
}
@Test
- void testSendBySessionFailure() {
+ void testSendBySessionFailure() throws Exception {
final RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
when(session.getAsyncRemote()).thenReturn(async);
when(session.isOpen()).thenReturn(true);
@@ -562,10 +563,87 @@ public final class WebsocketCollectorTest {
}).when(async).sendText(anyString(), any(SendHandler.class));
WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"fail-msg", DataEventTypeEnum.CREATE);
verify(async, times(1)).sendText(eq("fail-msg"),
any(SendHandler.class));
+ // a failed send closes the session so the gateway reconnects and
resynchronizes
+ verify(session, times(1)).close();
+ assertEquals(0, sendQueueSize());
websocketCollector.onClose(session);
}
+ @Test
+ void testCloseCancelsInFlightWatchdog() {
+ WebsocketCollector.setSendTimeoutMillis(60000L);
+ final RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
+ when(session.getAsyncRemote()).thenReturn(async);
+ websocketCollector.onOpen(session);
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"stalled-message", DataEventTypeEnum.CREATE);
+ Map<?, ?> queues = (Map<?, ?>)
ReflectionTestUtils.getField(WebsocketCollector.class, "SESSION_SEND_QUEUES");
+ Object queue = java.util.Objects.requireNonNull(queues).get(session);
+ java.util.concurrent.ScheduledFuture<?> future =
(java.util.concurrent.ScheduledFuture<?>)
+ ReflectionTestUtils.getField(queue, "timeoutFuture");
+ org.junit.jupiter.api.Assertions.assertNotNull(future);
+ assertFalse(future.isCancelled());
+ websocketCollector.onClose(session);
+ assertTrue(future.isCancelled());
+ assertNull(ReflectionTestUtils.getField(queue, "timeoutFuture"));
+ WebsocketCollector.resetSendGuards();
+ }
+
+ @Test
+ void testMissingSendCallbackTimesOutAndClosesSession() throws Exception {
+ WebsocketCollector.setSendTimeoutMillis(150);
+ final RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
+ when(session.getAsyncRemote()).thenReturn(async);
+ // container never invokes the send handler — the stalled-connection
case
+ doAnswer(invocation -> null).when(async).sendText(anyString(),
any(SendHandler.class));
+ websocketCollector.onOpen(session);
+
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"stuck-message", DataEventTypeEnum.CREATE);
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"second-message", DataEventTypeEnum.CREATE);
+ // only the first message reaches sendText; the second stays queued
+ verify(async, times(1)).sendText(eq("stuck-message"),
any(SendHandler.class));
+ verify(async, never()).sendText(eq("second-message"),
any(SendHandler.class));
+
+ // the watchdog closes the broken session within a bounded time
+ waitUntil(() -> getSessionSetSize() == 0L);
+ verify(session, times(1)).close();
+ assertEquals(0, sendQueueSize());
+ }
+
+ @Test
+ void testSynchronousSendExceptionClosesSession() throws Exception {
+ final RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
+ when(session.getAsyncRemote()).thenReturn(async);
+ when(session.isOpen()).thenReturn(true);
+ websocketCollector.onOpen(session);
+
+ doAnswer(invocation -> {
+ throw new IllegalStateException("broken pipe");
+ }).when(async).sendText(anyString(), any(SendHandler.class));
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"boom-message", DataEventTypeEnum.CREATE);
+
+ verify(session, times(1)).close();
+ assertEquals(0, sendQueueSize());
+ }
+
+ @Test
+ void testQueueOverflowClosesSession() throws Exception {
+ WebsocketCollector.setMaxQueuedMessages(2);
+ final RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
+ when(session.getAsyncRemote()).thenReturn(async);
+ doAnswer(invocation -> null).when(async).sendText(anyString(),
any(SendHandler.class));
+ websocketCollector.onOpen(session);
+
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID, "m1",
DataEventTypeEnum.CREATE);
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID, "m2",
DataEventTypeEnum.CREATE);
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID, "m3",
DataEventTypeEnum.CREATE);
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID, "m4",
DataEventTypeEnum.CREATE);
+
+ verify(async, times(1)).sendText(eq("m1"), any(SendHandler.class));
+ verify(session, times(1)).close();
+ assertEquals(0, sendQueueSize());
+ }
+
@Test
void testSendDoesNotWaitForOtherSession() {
Session anotherSession = mock(Session.class);
@@ -640,4 +718,11 @@ public final class WebsocketCollectorTest {
private Session getSession() {
return (Session) ThreadLocalUtils.get("sessionKey");
}
+
+ private void waitUntil(final java.util.function.Supplier<Boolean>
condition) throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 3000L;
+ while (!condition.get() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(20L);
+ }
+ }
}
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginTest.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginTest.java
index 66f6ee5bbe..1cedb6e687 100644
---
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginTest.java
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginTest.java
@@ -30,6 +30,8 @@ import
org.apache.shenyu.e2e.engine.scenario.specification.CaseSpec;
import org.apache.shenyu.e2e.enums.ServiceTypeEnum;
import org.apache.shenyu.e2e.model.ResourcesData;
import org.apache.shenyu.e2e.model.data.BindingData;
+import org.apache.shenyu.e2e.model.data.RuleCacheData;
+import org.apache.shenyu.e2e.model.data.SelectorCacheData;
import org.apache.shenyu.e2e.model.response.SelectorDTO;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -37,12 +39,15 @@ import org.junit.jupiter.api.BeforeEach;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.stream.Collectors;
import static
org.apache.shenyu.e2e.constant.Constants.SYS_DEFAULT_NAMESPACE_NAMESPACE_ID;
+import static org.awaitility.Awaitility.await;
@ShenYuTest(environments = {
@ShenYuTest.Environment(
@@ -75,12 +80,13 @@ public class DividePluginTest {
spec.getChecker().check(gateway);
ResourcesData resources = spec.getResources();
+ List<String> ruleIds = Lists.newArrayList();
for (ResourcesData.Resource res : resources.getResources()) {
SelectorDTO dto = client.create(res.getSelector());
selectorIds.add(dto.getId());
res.getRules().forEach(rule -> {
rule.setSelectorId(dto.getId());
- client.create(rule);
+ ruleIds.add(client.create(rule).getId());
});
BindingData bindingData = res.getBindingData();
if (Objects.nonNull(bindingData)) {
@@ -90,6 +96,10 @@ public class DividePluginTest {
}
}
+ // Admin creation does not imply that the gateway has received the
logging rules.
+ if (!selectorIds.isEmpty()) {
+ waitForLoggingRules(gateway, selectorIds, ruleIds);
+ }
spec.getWaiting().waitFor(gateway);
}
@@ -100,6 +110,13 @@ public class DividePluginTest {
selectorIds = Lists.newArrayList();
}
+ static void waitForLoggingRules(final GatewayClient gateway, final
List<String> selectors, final List<String> rules) {
+ await().alias("RocketMQ selectors and rules synchronized to gateway")
+ .atMost(Duration.ofSeconds(30))
+ .until(() ->
gateway.getSelectorCache().stream().map(SelectorCacheData::getId).collect(Collectors.toSet()).containsAll(selectors)
+ &&
gateway.getRuleCache().stream().map(RuleCacheData::getId).collect(Collectors.toSet()).containsAll(rules));
+ }
+
@BeforeAll
void setup(final AdminClient adminClient, final GatewayClient
gatewayClient) throws Exception {
adminClient.login();
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/LoggingRuleSyncTest.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/LoggingRuleSyncTest.java
new file mode 100644
index 0000000000..b0403346f9
--- /dev/null
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/LoggingRuleSyncTest.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.e2e.testcase.logging.rocketmq;
+
+import org.apache.shenyu.e2e.client.gateway.GatewayClient;
+import org.apache.shenyu.e2e.model.data.RuleCacheData;
+import org.apache.shenyu.e2e.model.data.SelectorCacheData;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class LoggingRuleSyncTest {
+
+ @Test
+ void waitsForNewIdsEvenWhenCacheSizesMatch() {
+ AtomicInteger selectorReads = new AtomicInteger();
+ AtomicInteger ruleReads = new AtomicInteger();
+ GatewayClient gateway = new GatewayClient("sync-test", "gateway",
"http://localhost";, new Properties()) {
+ @Override
+ public List<SelectorCacheData> getSelectorCache() {
+ SelectorCacheData selector = new SelectorCacheData();
+ selector.setId(selectorReads.incrementAndGet() < 3 ?
"old-selector" : "new-selector");
+ return List.of(selector);
+ }
+
+ @Override
+ public List<RuleCacheData> getRuleCache() {
+ RuleCacheData rule = new RuleCacheData();
+ rule.setId(ruleReads.incrementAndGet() < 3 ? "old-rule" :
"new-rule");
+ return List.of(rule);
+ }
+ };
+
+ DividePluginTest.waitForLoggingRules(gateway, List.of("new-selector"),
List.of("new-rule"));
+
+ assertTrue(selectorReads.get() >= 3);
+ assertTrue(ruleReads.get() >= 3);
+ }
+
+ @Test
+ void acceptsAlreadySynchronizedRulesAlongsideUnrelatedData() {
+ GatewayClient gateway = new GatewayClient("sync-test", "gateway",
"http://localhost";, new Properties()) {
+ @Override
+ public List<SelectorCacheData> getSelectorCache() {
+ SelectorCacheData selector = new SelectorCacheData();
+ selector.setId("new-selector");
+ return List.of(selector, new SelectorCacheData());
+ }
+
+ @Override
+ public List<RuleCacheData> getRuleCache() {
+ RuleCacheData rule = new RuleCacheData();
+ rule.setId("new-rule");
+ return List.of(rule, new RuleCacheData());
+ }
+ };
+
+ DividePluginTest.waitForLoggingRules(gateway, List.of("new-selector"),
List.of("new-rule"));
+ }
+}
