Copilot commented on code in PR #7672:
URL: https://github.com/apache/shenyu/pull/7672#discussion_r4291630700
##########
shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java:
##########
@@ -160,26 +162,59 @@ 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();
+ final Set<String> selectorIds =
dataList.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 (!dataList.isEmpty()) {
+ cache.refreshSelectorData(dataList);
+ }
+ if (!stale.isEmpty()) {
+ cache.removeSelectDataBatch(stale);
Review Comment:
Both snapshot methods delete all stale rows before invoking any handlers. If
an incoming handler throws during InitialSyncApplication, execution stops
before the stale removal callbacks and match-cache invalidation. A retry cannot
rediscover those deleted rows, so cleanup remains skipped even after a
successful retry—for example, Divide's selector upstream entries and gRPC's
rule handles remain cached. Previously, an incoming-handler failure occurred
before stale deletion. Keep pending removals retryable, or finish the cleanup
and invalidation before propagating the failure. Add failure-then-retry tests
for both snapshot methods.
##########
shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java:
##########
@@ -160,26 +162,59 @@ 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();
+ final Set<String> selectorIds =
dataList.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 (!dataList.isEmpty()) {
+ cache.refreshSelectorData(dataList);
+ }
+ if (!stale.isEmpty()) {
+ cache.removeSelectDataBatch(stale);
+ }
+ // Preserve per-item plugin handler callbacks and the initial-sync
failure semantics
+ // that the previous per-row subscribe path provided.
+ dataList.forEach(data -> runGuarded(data, () ->
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));
+ }
+ }));
+ invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(dataList,
stale, SelectorData::getPluginName));
}
@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();
+ final Set<String> ruleIds =
dataList.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 (!dataList.isEmpty()) {
+ cache.refreshRuleData(dataList);
+ }
+ if (!stale.isEmpty()) {
+ cache.removeRuleDataBatch(stale);
+ }
+ dataList.forEach(data -> runGuarded(data, () ->
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));
+ }
+ }));
+ invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(dataList,
stale, RuleData::getPluginName));
Review Comment:
A stale rule can have a blank pluginName because BaseDataCache keys rules by
selectorId. The batch removes that rule, but collectAffectedPlugins drops its
blank name, leaving positive matches for its id under the former plugin
untouched. The previous unRuleSubscribe path used the id-based cross-plugin
invalidation fallback. Preserve that fallback for stale rules whose plugin name
is unavailable.
##########
shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java:
##########
@@ -160,26 +162,59 @@ 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();
+ final Set<String> selectorIds =
dataList.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 (!dataList.isEmpty()) {
+ cache.refreshSelectorData(dataList);
Review Comment:
Both snapshot paths now call helpers that copy the entire selector/rule map
and later replace its reference. An incremental subscribe or unsubscribe that
completes after the copy can be overwritten at publication, even in an
unrelated plugin or selector bucket. The previous per-row snapshot writes used
live-map computes and preserved those unrelated updates; the existing
refresh-path race does not make this new usage safe. Use grouped computes on
the live map, or coordinate publication with all writers. Add deterministic
snapshot-versus-incremental tests for both selectors and rules.
--
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]
