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 628c8351cb fix(sync): retain gateway cache during namespace snapshots
(#5580) (#7416)
628c8351cb is described below
commit 628c8351cbde2a43116a8695184470bcca5d7053
Author: JerryčAI <[email protected]>
AuthorDate: Sun Oct 11 23:20:19 2026 +0800
fix(sync): retain gateway cache during namespace snapshots (#5580) (#7416)
---
.../base/cache/CommonPluginDataSubscriber.java | 36 +++++++++++++
.../base/cache/CommonPluginDataSubscriberTest.java | 61 ++++++++++++++++++++++
.../shenyu/sync/data/api/PluginDataSubscriber.java | 33 ++++++++++++
.../data/websocket/handler/PluginDataHandler.java | 3 +-
.../data/websocket/handler/RuleDataHandler.java | 3 +-
.../websocket/handler/SelectorDataHandler.java | 3 +-
.../handler/WebsocketDataHandlerTest.java | 16 +++---
7 files changed, 142 insertions(+), 13 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
index 181b95dad4..c72c39e6e0 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
@@ -40,6 +40,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -146,6 +147,41 @@ public class CommonPluginDataSubscriber implements
PluginDataSubscriber {
stale.forEach(this::unRuleSubscribe);
}
+ @Override
+ public void applyPluginDataSnapshot(final String namespaceId, final
List<PluginData> dataList) {
+ Set<String> pluginNames =
dataList.stream().map(PluginData::getName).collect(Collectors.toSet());
+ List<PluginData> stale =
BaseDataCache.getInstance().getPluginMap().values().stream()
+ .filter(data -> namespaceId.equals(data.getNamespaceId()))
+ .filter(data -> !pluginNames.contains(data.getName()))
+ .collect(Collectors.toList());
+ dataList.forEach(this::onSubscribe);
+ stale.forEach(this::unSubscribe);
+ }
+
+ @Override
+ public void applySelectorDataSnapshot(final String namespaceId, final
List<SelectorData> dataList) {
+ Set<String> selectorIds =
dataList.stream().map(SelectorData::getId).collect(Collectors.toSet());
+ List<SelectorData> stale =
BaseDataCache.getInstance().getSelectorMap().values().stream()
+ .flatMap(List::stream)
+ .filter(data -> namespaceId.equals(data.getNamespaceId()))
+ .filter(data -> !selectorIds.contains(data.getId()))
+ .collect(Collectors.toList());
+ dataList.forEach(this::onSelectorSubscribe);
+ stale.forEach(this::unSelectorSubscribe);
+ }
+
+ @Override
+ public void applyRuleDataSnapshot(final String namespaceId, final
List<RuleData> dataList) {
+ Set<String> ruleIds =
dataList.stream().map(RuleData::getId).collect(Collectors.toSet());
+ List<RuleData> stale =
BaseDataCache.getInstance().getRuleMap().values().stream()
+ .flatMap(List::stream)
+ .filter(data -> namespaceId.equals(data.getNamespaceId()))
+ .filter(data -> !ruleIds.contains(data.getId()))
+ .collect(Collectors.toList());
+ dataList.forEach(this::onRuleSubscribe);
+ stale.forEach(this::unRuleSubscribe);
+ }
+
@Override
public void refreshPluginDataAll() {
BaseDataCache.getInstance().cleanPluginData();
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
index cedf095d37..8487c594c4 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
@@ -109,6 +109,67 @@ public final class CommonPluginDataSubscriberTest {
assertEquals(pluginData,
baseDataCache.obtainPluginData(pluginData.getName()));
}
+ @Test
+ void testPluginSnapshotKeepsExistingEntryUntilReplacementIsApplied() {
+ final PluginData oldPlugin =
PluginData.builder().id("divide-id").name("divide").config("old")
+ .namespaceId("namespace-a").build();
+ final PluginData replacement =
PluginData.builder().id("divide-id").name("divide").config("new")
+ .namespaceId("namespace-a").build();
+ final PluginData stale =
PluginData.builder().id("removed-id").name("removed").namespaceId("namespace-a").build();
+ final PluginData otherNamespace =
PluginData.builder().id("other-id").name("other").namespaceId("namespace-b").build();
+ baseDataCache.cachePluginData(oldPlugin);
+ baseDataCache.cachePluginData(stale);
+ baseDataCache.cachePluginData(otherNamespace);
+ doAnswer(invocation -> {
+ assertEquals(oldPlugin, baseDataCache.obtainPluginData("divide"));
+ return null;
+ }).when(handler).handlerPlugin(replacement);
+
+ commonPluginDataSubscriber.applyPluginDataSnapshot("namespace-a",
List.of(replacement));
+
+ assertEquals(replacement, baseDataCache.obtainPluginData("divide"));
+ assertNull(baseDataCache.obtainPluginData("removed"));
+ assertEquals(otherNamespace, baseDataCache.obtainPluginData("other"));
+ }
+
+ @Test
+ void testSelectorSnapshotReplacesRowsAndRemovesOnlyStaleNamespaceRows() {
+ final SelectorData oldSelector =
SelectorData.builder().id("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final SelectorData replacement =
SelectorData.builder().id("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ final SelectorData stale =
SelectorData.builder().id("stale-selector").pluginName("divide")
+ .namespaceId("namespace-a").sort(3).build();
+ final SelectorData otherNamespace =
SelectorData.builder().id("other-selector").pluginName("divide")
+ .namespaceId("namespace-b").sort(4).build();
+ baseDataCache.cacheSelectData(oldSelector);
+ baseDataCache.cacheSelectData(stale);
+ baseDataCache.cacheSelectData(otherNamespace);
+
+ commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(replacement));
+
+ assertEquals(List.of(replacement, otherNamespace),
baseDataCache.obtainSelectorData("divide"));
+ }
+
+ @Test
+ void testRuleSnapshotReplacesRowsAndRemovesOnlyStaleNamespaceRows() {
+ final RuleData oldRule =
RuleData.builder().id("rule-a").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final RuleData replacement =
RuleData.builder().id("rule-a").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ final RuleData stale =
RuleData.builder().id("stale-rule").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(3).build();
+ final RuleData otherNamespace =
RuleData.builder().id("other-rule").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-b").sort(4).build();
+ baseDataCache.cacheRuleData(oldRule);
+ baseDataCache.cacheRuleData(stale);
+ baseDataCache.cacheRuleData(otherNamespace);
+
+ commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a",
List.of(replacement));
+
+ assertEquals(List.of(replacement, otherNamespace),
baseDataCache.obtainRuleData("selector-a"));
+ }
+
@Test
void
testInitialSyncPropagatesHandlerFailureWithoutChangingLegacyBehavior() {
PluginDataHandler handler = mock(PluginDataHandler.class);
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-api/src/main/java/org/apache/shenyu/sync/data/api/PluginDataSubscriber.java
b/shenyu-sync-data-center/shenyu-sync-data-api/src/main/java/org/apache/shenyu/sync/data/api/PluginDataSubscriber.java
index 10026fdb3d..4bf3418607 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-api/src/main/java/org/apache/shenyu/sync/data/api/PluginDataSubscriber.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-api/src/main/java/org/apache/shenyu/sync/data/api/PluginDataSubscriber.java
@@ -177,4 +177,37 @@ public interface PluginDataSubscriber {
throw new UnsupportedOperationException("Namespace-scoped rule
snapshots are not supported");
}
+ /**
+ * Apply an authoritative plugin snapshot while retaining cached rows
present in the snapshot.
+ *
+ * @param namespaceId namespace to reconcile
+ * @param dataList complete plugin snapshot for the namespace
+ */
+ default void applyPluginDataSnapshot(final String namespaceId, final
List<PluginData> dataList) {
+ refreshPluginDataNamespace(namespaceId);
+ dataList.forEach(this::onSubscribe);
+ }
+
+ /**
+ * Apply an authoritative selector snapshot while retaining cached rows
present in the snapshot.
+ *
+ * @param namespaceId namespace to reconcile
+ * @param dataList complete selector snapshot for the namespace
+ */
+ default void applySelectorDataSnapshot(final String namespaceId, final
List<SelectorData> dataList) {
+ refreshSelectorDataNamespace(namespaceId);
+ dataList.forEach(this::onSelectorSubscribe);
+ }
+
+ /**
+ * Apply an authoritative rule snapshot while retaining cached rows
present in the snapshot.
+ *
+ * @param namespaceId namespace to reconcile
+ * @param dataList complete rule snapshot for the namespace
+ */
+ default void applyRuleDataSnapshot(final String namespaceId, final
List<RuleData> dataList) {
+ refreshRuleDataNamespace(namespaceId);
+ dataList.forEach(this::onRuleSubscribe);
+ }
+
}
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/PluginDataHandler.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/PluginDataHandler.java
index 055c95009d..0f0e38090c 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/PluginDataHandler.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/PluginDataHandler.java
@@ -50,8 +50,7 @@ public class PluginDataHandler extends
AbstractDataHandler<PluginData> {
if (dataList.stream().anyMatch(data ->
!namespaceId.equals(data.getNamespaceId()))) {
throw new IllegalArgumentException("Snapshot row namespace does
not match the connection");
}
- pluginDataSubscriber.refreshPluginDataNamespace(namespaceId);
- doUpdate(dataList);
+ pluginDataSubscriber.applyPluginDataSnapshot(namespaceId, dataList);
}
@Override
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/RuleDataHandler.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/RuleDataHandler.java
index 0fbdbed978..651b50dda3 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/RuleDataHandler.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/RuleDataHandler.java
@@ -49,8 +49,7 @@ public class RuleDataHandler extends
AbstractDataHandler<RuleData> {
if (dataList.stream().anyMatch(data ->
!namespaceId.equals(data.getNamespaceId()))) {
throw new IllegalArgumentException("Snapshot row namespace does
not match the connection");
}
- pluginDataSubscriber.refreshRuleDataNamespace(namespaceId);
- doUpdate(dataList);
+ pluginDataSubscriber.applyRuleDataSnapshot(namespaceId, dataList);
}
@Override
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/SelectorDataHandler.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/SelectorDataHandler.java
index 3b142c49c6..840d5708c9 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/SelectorDataHandler.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/handler/SelectorDataHandler.java
@@ -50,8 +50,7 @@ public class SelectorDataHandler extends
AbstractDataHandler<SelectorData> {
if (dataList.stream().anyMatch(data ->
!namespaceId.equals(data.getNamespaceId()))) {
throw new IllegalArgumentException("Snapshot row namespace does
not match the connection");
}
- pluginDataSubscriber.refreshSelectorDataNamespace(namespaceId);
- doUpdate(dataList);
+ pluginDataSubscriber.applySelectorDataSnapshot(namespaceId, dataList);
}
@Override
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/handler/WebsocketDataHandlerTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/handler/WebsocketDataHandlerTest.java
index 1b69ddef9f..12ecff787e 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/handler/WebsocketDataHandlerTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/handler/WebsocketDataHandlerTest.java
@@ -41,6 +41,7 @@ import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
import org.apache.shenyu.sync.data.api.ProxySelectorDataSubscriber;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.springframework.context.ApplicationEventPublisher;
import org.mockito.Mockito;
public final class WebsocketDataHandlerTest {
@@ -156,7 +157,7 @@ public final class WebsocketDataHandlerTest {
@Test
public void testEmptySnapshotClearsOnlyItsGroup() {
websocketDataHandler.snapshot(ConfigGroupEnum.RULE, "[]",
"namespace-a", "namespace-a");
- verify(pluginDataSubscriber).refreshRuleDataNamespace("namespace-a");
+ verify(pluginDataSubscriber).applyRuleDataSnapshot("namespace-a",
Collections.emptyList());
Mockito.verifyNoMoreInteractions(pluginDataSubscriber);
}
@@ -177,15 +178,14 @@ public final class WebsocketDataHandlerTest {
@Test
public void testSnapshotReplacesStalePluginBeforeSubscribing() {
websocketDataHandler.snapshot(ConfigGroupEnum.PLUGIN, getJson(),
"namespace-a", "namespace-a");
- org.mockito.InOrder order = Mockito.inOrder(pluginDataSubscriber);
-
order.verify(pluginDataSubscriber).refreshPluginDataNamespace("namespace-a");
-
order.verify(pluginDataSubscriber).onSubscribe(Mockito.any(PluginData.class));
+
verify(pluginDataSubscriber).applyPluginDataSnapshot(Mockito.eq("namespace-a"),
Mockito.anyList());
}
@Test
public void testEmptySnapshotsRemoveStaleGatewayCache() {
BaseDataCache cache = BaseDataCache.getInstance();
PluginDataSubscriber subscriber = new
CommonPluginDataSubscriber(Collections.emptyList(),
+ mock(ApplicationEventPublisher.class),
new ShenyuConfig.SelectorMatchCache(), new
ShenyuConfig.RuleMatchCache());
WebsocketDataHandler handler = new WebsocketDataHandler(subscriber,
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
@@ -229,6 +229,7 @@ public final class WebsocketDataHandlerTest {
public void testEmptyNamespaceSnapshotPreservesOtherNamespaces() {
BaseDataCache cache = BaseDataCache.getInstance();
PluginDataSubscriber subscriber = new
CommonPluginDataSubscriber(Collections.emptyList(),
+ mock(ApplicationEventPublisher.class),
new ShenyuConfig.SelectorMatchCache(), new
ShenyuConfig.RuleMatchCache());
WebsocketDataHandler handler = new WebsocketDataHandler(subscriber,
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
@@ -261,14 +262,15 @@ public final class WebsocketDataHandlerTest {
public void
testNonEmptySnapshotRemovesMissingRulesAndPreservesOtherNamespace() {
BaseDataCache cache = BaseDataCache.getInstance();
PluginDataSubscriber subscriber = new
CommonPluginDataSubscriber(Collections.emptyList(),
+ mock(ApplicationEventPublisher.class),
new ShenyuConfig.SelectorMatchCache(), new
ShenyuConfig.RuleMatchCache());
WebsocketDataHandler handler = new WebsocketDataHandler(subscriber,
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
- RuleData stale =
RuleData.builder().id("stale-rule").selectorId("selector-a")
+ RuleData stale =
RuleData.builder().id("stale-rule").selectorId("selector-a").pluginName("divide")
.namespaceId("namespace-a").sort(1).build();
- RuleData replacement =
RuleData.builder().id("replacement-rule").selectorId("selector-a")
+ RuleData replacement =
RuleData.builder().id("replacement-rule").selectorId("selector-a").pluginName("divide")
.namespaceId("namespace-a").sort(2).build();
- RuleData other =
RuleData.builder().id("other-rule").selectorId("selector-b")
+ RuleData other =
RuleData.builder().id("other-rule").selectorId("selector-b").pluginName("divide")
.namespaceId("namespace-b").sort(1).build();
cache.cacheRuleData(stale);
cache.cacheRuleData(other);
