diff --git a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
index fed4420289db..fc67d1143a31 100644
--- a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
+++ b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
@@ -25,6 +25,8 @@
 
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -179,6 +181,44 @@ public void cleanSelectorData() {
     public void cleanSelectorDataSelf(final List<SelectorData> selectorDataList) {
         selectorDataList.forEach(this::removeSelectData);
     }
+
+    /**
+     * Remove a batch of selectors with a single merge per affected plugin bucket.
+     * Selectors without a plugin name are matched against every bucket, mirroring
+     * {@link #removeSelectData(SelectorData)}.
+     *
+     * @param selectorDataList the selectors to remove
+     */
+    void removeSelectDataBatch(final List<SelectorData> selectorDataList) {
+        if (selectorDataList.isEmpty()) {
+            return;
+        }
+        final Map<String, Set<String>> idsByPlugin = new HashMap<>();
+        final Set<String> danglingIds = new HashSet<>();
+        for (SelectorData data : selectorDataList) {
+            if (Objects.isNull(data) || Objects.isNull(data.getId())) {
+                continue;
+            }
+            if (StringUtils.isBlank(data.getPluginName())) {
+                danglingIds.add(data.getId());
+            } else {
+                idsByPlugin.computeIfAbsent(data.getPluginName(), key -> new HashSet<>()).add(data.getId());
+            }
+        }
+        if (!danglingIds.isEmpty()) {
+            selectorMap.keySet().forEach(pluginName -> removeSelectDataByIds(pluginName, danglingIds));
+        }
+        idsByPlugin.forEach(this::removeSelectDataByIds);
+    }
+
+    private void removeSelectDataByIds(final String pluginName, final Set<String> selectorIds) {
+        selectorMap.computeIfPresent(pluginName, (key, value) -> {
+            final List<SelectorData> result = value.stream()
+                    .filter(selector -> !selectorIds.contains(selector.getId()))
+                    .collect(Collectors.toList());
+            return result.isEmpty() ? null : List.copyOf(result);
+        });
+    }
     
     /**
      * Obtain selector data list list.
@@ -239,6 +279,30 @@ public void cleanRuleData() {
     public void cleanRuleDataSelf(final List<RuleData> ruleDataList) {
         ruleDataList.forEach(this::removeRuleData);
     }
+
+    /**
+     * Remove a batch of rules with a single merge per affected selector bucket.
+     *
+     * @param ruleDataList the rules to remove
+     */
+    void removeRuleDataBatch(final List<RuleData> ruleDataList) {
+        if (ruleDataList.isEmpty()) {
+            return;
+        }
+        final Map<String, Set<String>> idsBySelector = new HashMap<>();
+        for (RuleData data : ruleDataList) {
+            if (Objects.isNull(data) || Objects.isNull(data.getId()) || Objects.isNull(data.getSelectorId())) {
+                continue;
+            }
+            idsBySelector.computeIfAbsent(data.getSelectorId(), key -> new HashSet<>()).add(data.getId());
+        }
+        idsBySelector.forEach((selectorId, ruleIds) -> ruleMap.computeIfPresent(selectorId, (key, value) -> {
+            final List<RuleData> result = value.stream()
+                    .filter(rule -> !ruleIds.contains(rule.getId()))
+                    .collect(Collectors.toList());
+            return result.isEmpty() ? null : List.copyOf(result);
+        }));
+    }
     
     /**
      * Obtain rule data list list.
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 c72c39e6e076..9c171358b41b 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
@@ -36,12 +36,16 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.lang.NonNull;
 
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * The type Common plugin data subscriber.
@@ -160,26 +164,100 @@ public void applyPluginDataSnapshot(final String namespaceId, final List<PluginD
 
     @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()
+        final BaseDataCache cache = BaseDataCache.getInstance();
+        // A row without a plugin name yields a null grouping key in refreshSelectorData, whose compute on a
+        // concurrent map throws and would abort the whole snapshot (and, over a websocket, loop resyncs).
+        // The previous per-row subscribe path isolated such a row; skip it up front so valid rows still converge.
+        final List<SelectorData> incoming = new ArrayList<>(dataList.size());
+        for (SelectorData data : dataList) {
+            if (Objects.nonNull(data) && Objects.nonNull(data.getPluginName())) {
+                incoming.add(data);
+            } else {
+                LOG.warn("skip malformed selector row without plugin name in snapshot for namespace [{}]", namespaceId);
+            }
+        }
+        final Set<String> selectorIds = incoming.stream().map(SelectorData::getId).collect(Collectors.toSet());
+        final List<SelectorData> stale = cache.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);
+        // Merge and sort once per affected plugin bucket instead of per selector row.
+        if (!incoming.isEmpty()) {
+            cache.refreshSelectorData(incoming);
+        }
+        if (!stale.isEmpty()) {
+            cache.removeSelectDataBatch(stale);
+        }
+        try {
+            // Preserve per-item plugin handler callbacks and the initial-sync failure semantics
+            // that the previous per-row subscribe path provided.
+            incoming.forEach(data -> runGuarded(data, () -> {
+                if (StringUtils.isNotBlank(data.getPluginName())) {
+                    Optional.ofNullable(handlerMap.get(data.getPluginName()))
+                            .ifPresent(handler -> handler.handlerSelector(data));
+                }
+            }));
+            stale.forEach(data -> runGuarded(data, () -> {
+                // the concurrent handler map rejects null keys, and a dangling selector carries no plugin name
+                if (StringUtils.isNotBlank(data.getPluginName())) {
+                    Optional.ofNullable(handlerMap.get(data.getPluginName()))
+                            .ifPresent(handler -> handler.removeSelector(data));
+                }
+            }));
+        } finally {
+            // Invalidate even if a handler aborted during initial sync: the stale rows are already gone from
+            // BaseDataCache above, so a retry cannot rediscover them and their match cache would stay stale.
+            invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(incoming, stale, SelectorData::getPluginName));
+            invalidateDanglingSelectorMatchCache(incoming, stale);
+        }
     }
 
     @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()
+        final BaseDataCache cache = BaseDataCache.getInstance();
+        // A row without a selector id yields a null grouping key in refreshRuleData, whose compute on a
+        // concurrent map throws and would abort the whole snapshot; skip it up front so valid rows still converge.
+        final List<RuleData> incoming = new ArrayList<>(dataList.size());
+        for (RuleData data : dataList) {
+            if (Objects.nonNull(data) && Objects.nonNull(data.getSelectorId())) {
+                incoming.add(data);
+            } else {
+                LOG.warn("skip malformed rule row without selector id in snapshot for namespace [{}]", namespaceId);
+            }
+        }
+        final Set<String> ruleIds = incoming.stream().map(RuleData::getId).collect(Collectors.toSet());
+        final List<RuleData> stale = cache.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);
+        // Merge and sort once per affected selector bucket instead of per rule row.
+        if (!incoming.isEmpty()) {
+            cache.refreshRuleData(incoming);
+        }
+        if (!stale.isEmpty()) {
+            cache.removeRuleDataBatch(stale);
+        }
+        try {
+            incoming.forEach(data -> runGuarded(data, () -> {
+                if (StringUtils.isNotBlank(data.getPluginName())) {
+                    Optional.ofNullable(handlerMap.get(data.getPluginName()))
+                            .ifPresent(handler -> handler.handlerRule(data));
+                }
+            }));
+            stale.forEach(data -> runGuarded(data, () -> {
+                if (StringUtils.isNotBlank(data.getPluginName())) {
+                    Optional.ofNullable(handlerMap.get(data.getPluginName()))
+                            .ifPresent(handler -> handler.removeRule(data));
+                }
+            }));
+        } finally {
+            // Invalidate even if a handler aborted during initial sync, so a retry is not left with stale
+            // match-cache entries for rows already removed from BaseDataCache above.
+            invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(incoming, stale, RuleData::getPluginName));
+            invalidateDanglingRuleMatchCache(incoming, stale);
+        }
     }
 
     @Override
@@ -402,6 +480,115 @@ private void invalidateRuleMatchCache(final RuleData ruleData) {
         }
     }
 
+    /**
+     * Run a per-row callback for a snapshot apply, preserving the exception semantics of
+     * {@link #subscribeDataHandler}: during initial sync failures abort the snapshot, otherwise
+     * they are logged so subsequent rows still converge.
+     */
+    private <T> void runGuarded(final T data, final Runnable action) {
+        try {
+            action.run();
+        } catch (Exception e) {
+            if (InitialSyncApplication.isActive()) {
+                throw new IllegalStateException("Initial configuration application failed", e);
+            }
+            LOG.error("snapshot handler error, classData: {}", JsonUtils.toJson(data), e);
+        }
+    }
+
+    private <T> Set<String> collectAffectedPlugins(final List<T> incoming, final List<T> stale,
+                                                   final Function<T, String> pluginNameOf) {
+        final Set<String> plugins = new HashSet<>();
+        incoming.forEach(data -> {
+            if (Objects.isNull(data)) {
+                return;
+            }
+            final String name = pluginNameOf.apply(data);
+            if (StringUtils.isNotBlank(name)) {
+                plugins.add(name);
+            }
+        });
+        stale.forEach(data -> {
+            if (Objects.isNull(data)) {
+                return;
+            }
+            final String name = pluginNameOf.apply(data);
+            if (StringUtils.isNotBlank(name)) {
+                plugins.add(name);
+            }
+        });
+        return plugins;
+    }
+
+    /**
+     * Mirror master's id-based cross-bucket sweep for dangling selectors that carry no plugin name:
+     * {@link #collectAffectedPlugins} drops blank names, so without this a deleted dangling selector would
+     * leave its cached matches (which may live under any plugin bucket) behind. Delegates to the blank-name
+     * branch of {@link MatchDataCache#removeSelectorData(String, String)} and
+     * {@link MatchDataCache#removeRuleDataBySelector(String, String)}.
+     */
+    private void invalidateDanglingSelectorMatchCache(final List<SelectorData> incoming, final List<SelectorData> stale) {
+        if (!selectorMatchConfig.getCache().getEnabled() && !ruleMatchCacheConfig.getCache().getEnabled()) {
+            return;
+        }
+        final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+        Stream.concat(incoming.stream(), stale.stream())
+                .filter(Objects::nonNull)
+                .filter(data -> StringUtils.isBlank(data.getPluginName()))
+                .map(SelectorData::getId)
+                .filter(Objects::nonNull)
+                .forEach(id -> {
+                    if (selectorMatchConfig.getCache().getEnabled()) {
+                        matchDataCache.removeSelectorData(null, id);
+                    }
+                    if (ruleMatchCacheConfig.getCache().getEnabled()) {
+                        matchDataCache.removeRuleDataBySelector(null, id);
+                    }
+                });
+    }
+
+    /**
+     * Rule counterpart of {@link #invalidateDanglingSelectorMatchCache}: a stale rule keyed by selectorId can
+     * carry a blank plugin name, so its cached matches are swept across every bucket by rule id via the
+     * blank-name branch of {@link MatchDataCache#removeRuleData(String, String)}.
+     */
+    private void invalidateDanglingRuleMatchCache(final List<RuleData> incoming, final List<RuleData> stale) {
+        if (!ruleMatchCacheConfig.getCache().getEnabled()) {
+            return;
+        }
+        final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+        Stream.concat(incoming.stream(), stale.stream())
+                .filter(Objects::nonNull)
+                .filter(data -> StringUtils.isBlank(data.getPluginName()))
+                .map(RuleData::getId)
+                .filter(Objects::nonNull)
+                .forEach(id -> matchDataCache.removeRuleData(null, id));
+    }
+
+    private void invalidateSelectorMatchCacheByPlugins(final Set<String> pluginNames) {
+        if (pluginNames.isEmpty()) {
+            return;
+        }
+        final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+        for (String pluginName : pluginNames) {
+            if (selectorMatchConfig.getCache().getEnabled()) {
+                matchDataCache.removeSelectorData(pluginName);
+            }
+            if (ruleMatchCacheConfig.getCache().getEnabled()) {
+                // selector membership changed, so any cached rule match under this plugin may be stale
+                matchDataCache.removeRuleData(pluginName);
+            }
+        }
+    }
+
+    private void invalidateRuleMatchCacheByPlugins(final Set<String> pluginNames) {
+        if (pluginNames.isEmpty() || !ruleMatchCacheConfig.getCache().getEnabled()) {
+            return;
+        }
+        final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+        pluginNames.forEach(matchDataCache::removeRuleData);
+    }
+
     @Override
     public void onPluginRefresh(final List<PluginData> dataList) {
         if (CollectionUtils.isEmpty(dataList)) {
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 8487c594c463..bf49fbf85a70 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
@@ -584,4 +584,201 @@ private SelectorData selector(final String id, final int sort) {
     private RuleData rule(final String id, final int sort) {
         return RuleData.builder().id(id).pluginName("divide").selectorId("selector").sort(sort).build();
     }
+
+    @Test
+    public void selectorSnapshotBatchesMergesAndPreservesSortOrder() {
+        final SelectorData existing = SelectorData.builder().id("existing").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        baseDataCache.cacheSelectData(existing);
+        final SelectorData first = SelectorData.builder().id("s-3").pluginName("divide")
+                .namespaceId("namespace-a").sort(3).build();
+        final SelectorData second = SelectorData.builder().id("s-1").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        final SelectorData replacement = SelectorData.builder().id("existing").pluginName("divide")
+                .namespaceId("namespace-a").sort(4).build();
+
+        commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of(first, second, replacement));
+
+        // one merge+sort per plugin bucket: sorted by the sort key, replacement supersedes the existing row
+        assertEquals(List.of(second, first, replacement), baseDataCache.obtainSelectorData("divide"));
+        verify(handler).handlerSelector(first);
+        verify(handler).handlerSelector(second);
+        verify(handler).handlerSelector(replacement);
+    }
+
+    @Test
+    public void ruleSnapshotBatchesMergesAndPreservesSortOrder() {
+        final RuleData existing = RuleData.builder().id("existing").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        baseDataCache.cacheRuleData(existing);
+        final RuleData first = RuleData.builder().id("r-3").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(3).build();
+        final RuleData second = RuleData.builder().id("r-1").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        final RuleData replacement = RuleData.builder().id("existing").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(4).build();
+
+        commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a", List.of(first, second, replacement));
+
+        assertEquals(List.of(second, first, replacement), baseDataCache.obtainRuleData("selector-a"));
+        verify(handler).handlerRule(first);
+        verify(handler).handlerRule(second);
+        verify(handler).handlerRule(replacement);
+    }
+
+    @Test
+    public void selectorSnapshotInvokesRemoveSelectorForEachStaleRow() {
+        final SelectorData staleA = SelectorData.builder().id("stale-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        final SelectorData staleB = SelectorData.builder().id("stale-b").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        final SelectorData otherNamespace = SelectorData.builder().id("other").pluginName("divide")
+                .namespaceId("namespace-b").sort(3).build();
+        baseDataCache.cacheSelectData(staleA);
+        baseDataCache.cacheSelectData(staleB);
+        baseDataCache.cacheSelectData(otherNamespace);
+
+        commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of());
+
+        assertEquals(List.of(otherNamespace), baseDataCache.obtainSelectorData("divide"));
+        verify(handler).removeSelector(staleA);
+        verify(handler).removeSelector(staleB);
+    }
+
+    @Test
+    public void ruleSnapshotInvokesRemoveRuleForEachStaleRow() {
+        final RuleData staleA = RuleData.builder().id("stale-a").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        final RuleData staleB = RuleData.builder().id("stale-b").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        final RuleData otherNamespace = RuleData.builder().id("other").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-b").sort(3).build();
+        baseDataCache.cacheRuleData(staleA);
+        baseDataCache.cacheRuleData(staleB);
+        baseDataCache.cacheRuleData(otherNamespace);
+
+        commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a", List.of());
+
+        assertEquals(List.of(otherNamespace), baseDataCache.obtainRuleData("selector-a"));
+        verify(handler).removeRule(staleA);
+        verify(handler).removeRule(staleB);
+    }
+
+    @Test
+    public void snapshotCoalescesMatchCacheInvalidationPerAffectedPlugin() {
+        final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+        final RuleMatchCache ruleConfig = new RuleMatchCache();
+        selectorConfig.getCache().setEnabled(true);
+        ruleConfig.getCache().setEnabled(true);
+        commonPluginDataSubscriber = new CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig, ruleConfig);
+        final MatchDataCache matches = MatchDataCache.getInstance();
+        // Preload positive and negative cache entries for two plugins.
+        matches.cacheSelectorData("/divide", selector("s-1", 1), 100, 100);
+        matches.cacheSelectorData("/divide-empty", SelectorData.builder().pluginName("divide").build(), 100, 100);
+        matches.cacheRuleData("/divide", rule("r-1", 1), 100, 100);
+        final SelectorData otherPlugin = SelectorData.builder().id("s-other").pluginName("rewrite")
+                .namespaceId("namespace-a").sort(1).build();
+        matches.cacheSelectorData("/rewrite", otherPlugin, 100, 100);
+
+        // Snapshot only touches the "divide" plugin; the "rewrite" cache must remain.
+        final SelectorData incoming = SelectorData.builder().id("s-2").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of(incoming));
+
+        assertNull(matches.obtainSelectorData("divide", "/divide"));
+        assertNull(matches.obtainSelectorData("divide", "/divide-empty"));
+        assertNull(matches.obtainRuleData("divide", "/divide"));
+        assertNotNull(matches.obtainSelectorData("rewrite", "/rewrite"));
+    }
+
+    @Test
+    public void selectorSnapshotHandlerFailurePropagatesDuringInitialSync() {
+        final SelectorData incoming = SelectorData.builder().id("s-1").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        doThrow(new IllegalStateException("handler failed")).when(handler).handlerSelector(incoming);
+        assertThrows(IllegalStateException.class, () -> InitialSyncApplication.run(
+                () -> commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of(incoming))));
+        // cache write already published before the handler ran, mirroring the pre-batch behavior
+        assertEquals(List.of(incoming), baseDataCache.obtainSelectorData("divide"));
+    }
+
+    @Test
+    public void selectorSnapshotHandlerFailureIsLoggedOutsideInitialSync() {
+        final SelectorData first = SelectorData.builder().id("s-1").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        final SelectorData second = SelectorData.builder().id("s-2").pluginName("divide")
+                .namespaceId("namespace-a").sort(2).build();
+        doThrow(new IllegalStateException("handler failed")).when(handler).handlerSelector(first);
+        assertDoesNotThrow(() -> commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of(first, second)));
+        verify(handler).handlerSelector(second);
+        assertEquals(List.of(first, second), baseDataCache.obtainSelectorData("divide"));
+    }
+
+    @Test
+    public void danglingSelectorSnapshotSweepsMatchCacheAcrossBuckets() {
+        final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+        final RuleMatchCache ruleConfig = new RuleMatchCache();
+        selectorConfig.getCache().setEnabled(true);
+        ruleConfig.getCache().setEnabled(true);
+        commonPluginDataSubscriber = new CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig, ruleConfig);
+        final MatchDataCache matches = MatchDataCache.getInstance();
+
+        // a selector cached without a plugin name (dangling), plus match-cache entries under a real plugin bucket
+        final SelectorData dangling = SelectorData.builder().id("s-dangle").pluginName("").namespaceId("namespace-a").sort(1).build();
+        baseDataCache.cacheSelectData(dangling);
+        matches.cacheSelectorData("/dangle", SelectorData.builder().id("s-dangle").pluginName("divide").build(), 100, 100);
+        matches.cacheRuleData("/dangle", RuleData.builder().id("r-dangle").selectorId("s-dangle").pluginName("divide").build(), 100, 100);
+
+        // the snapshot drops the dangling selector; its blank name must still sweep every match-cache bucket by id
+        commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of());
+
+        assertNull(matches.obtainSelectorData("divide", "/dangle"));
+        assertNull(matches.obtainRuleData("divide", "/dangle"));
+    }
+
+    @Test
+    public void danglingRuleSnapshotSweepsMatchCacheAcrossBuckets() {
+        final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+        final RuleMatchCache ruleConfig = new RuleMatchCache();
+        selectorConfig.getCache().setEnabled(true);
+        ruleConfig.getCache().setEnabled(true);
+        commonPluginDataSubscriber = new CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig, ruleConfig);
+        final MatchDataCache matches = MatchDataCache.getInstance();
+
+        // a rule keyed by selectorId whose plugin name is blank, cached under a real plugin bucket
+        final RuleData dangling = RuleData.builder().id("r-dangle").selectorId("selector-a").pluginName("")
+                .namespaceId("namespace-a").sort(1).build();
+        baseDataCache.cacheRuleData(dangling);
+        matches.cacheRuleData("/dangle", RuleData.builder().id("r-dangle").selectorId("selector-a").pluginName("divide").build(), 100, 100);
+
+        commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a", List.of());
+
+        assertNull(matches.obtainRuleData("divide", "/dangle"));
+    }
+
+    @Test
+    public void malformedSelectorRowIsSkippedWithoutAbortingSnapshot() {
+        final SelectorData valid = SelectorData.builder().id("s-ok").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        // no plugin name: would produce a null grouping key and NPE inside the batch cache write
+        final SelectorData malformed = SelectorData.builder().id("s-bad").namespaceId("namespace-a").sort(2).build();
+
+        assertDoesNotThrow(() -> commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a", List.of(malformed, valid)));
+
+        assertEquals(List.of(valid), baseDataCache.obtainSelectorData("divide"));
+        verify(handler).handlerSelector(valid);
+    }
+
+    @Test
+    public void malformedRuleRowIsSkippedWithoutAbortingSnapshot() {
+        final RuleData valid = RuleData.builder().id("r-ok").selectorId("selector-a").pluginName("divide")
+                .namespaceId("namespace-a").sort(1).build();
+        // no selector id: would produce a null grouping key and NPE inside the batch cache write
+        final RuleData malformed = RuleData.builder().id("r-bad").pluginName("divide").namespaceId("namespace-a").sort(2).build();
+
+        assertDoesNotThrow(() -> commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a", List.of(malformed, valid)));
+
+        assertEquals(List.of(valid), baseDataCache.obtainRuleData("selector-a"));
+        verify(handler).handlerRule(valid);
+    }
 }