fix: evict stale upstream countMap entries in LeastActiveLoadBalance

[juicewcode]

Fixes #6891

• Fix the memory leak in LeastActiveLoadBalance.countMap. Previously entries were only added and never removed, so an upstream's entry lingered forever after it was taken out of the list, causing unbounded memory growth proportional to historical upstream churn and O(n) scan overhead on every selection. Now stale entries whose domains are absent from the current upstream list are evicted via countMap.keySet().retainAll(domainMap.keySet()) before selection.
• Added testRemoveStaleCountMapEntries to LeastActiveLoadBalanceTest. It registers two upstreams, removes one, and asserts the removed upstream's countMap entry is cleaned up while the remaining one is kept.

Make sure that:

• You have read the contribution guidelines.
• You submit test cases (unit or integration tests) that back your changes.
• Your local test passed ./mvnw clean install -Dmaven.javadoc.skip=true.

[nikhiln64]

Context

Thanks for tracking down the leak, the countMap really did grow without bound before this. One thing I would want to sort out before it lands. LeastActiveLoadBalance is a singleton (a bare @Join defaults to isSingleton true, and ExtensionLoader caches one instance per algorithm), so countMap at LeastActiveLoadBalance.java:37 is a single JVM wide map shared by every selector and rule that uses leastActive, keyed only by the upstream domain with no selector namespacing. The retainAll you have added runs on every doSelect against just the current request's upstream set, so two selectors with different upstream lists end up deleting each other's entries.

Example: If selector A serves {A1, A2} and selector B serves {B1, B2}, a request through A removes B1 and B2 from the map, and the next request through B removes A1 and A2 and re-adds B1 and B2 at Long.MIN_VALUE, discarding the counts B had accumulated. Under interleaved traffic that reset happens on nearly every request, so least active selection collapses toward near fixed selection and the map churns rather than just leaking.

There is also a smaller lost update where a computeIfPresent increment no ops if another thread's retainAll just removed that key.

Suggestion

At least gating the cleanup on the selector's own size mismatch rather than an unconditional retainAll, would fix the leak without touching other selector.

NOTE

Happy to be corrected if leastActive is only ever meant to serve one selector at a time, but nothing in the SPI seems to enforce that and RoundRobin assumes the opposite.

[juicewcode]

Thanks for the thorough review — you're right, and I've verified each point against the code.

Verified. LeastActiveLoadBalance is a bare @Join, so isSingleton defaults to true and
ExtensionLoader.getJoin("leastActive") returns one cached instance shared by every selector/rule
using leastActive. At the call site, DividePlugin fetches the upstream list per selector
(UpstreamCacheManager.findUpstreamListBySelectorId(selector.getId())), so the same singleton is
invoked with different upstream sets. My unconditional retainAll(domainMap.keySet()) therefore
really does evict another selector's live entries on every doSelect, and the lost computeIfPresent
increment is real — the original fix traded the leak for cross-selector count churn.

On the size-mismatch gate. I agree with the goal, but I think comparing sizes runs into a problem
here: countMap is a single map shared by all selectors and has no selector namespacing, so
countMap.size() is the union of everyone's domains and is not comparable to any single selector's
upstreamList.size(). In a multi-selector deployment the two sizes almost never match, so the gate
would stay open and the cross-selector deletion would still happen on nearly every request. A size
comparison also can't detect a member swap that keeps the list length unchanged (e.g. A1 replaced by
B1, size still 2), so stale entries could linger. In short, without namespacing, "the selector's own
size" isn't something we can actually measure — deriving one from the list would itself break whenever
a selector's first upstream changes.

Proposed change. I'd like to keep the single shared map and make eviction time-based instead,
like RoundRobinLoadBalancer's recycle logic but driven by a timestamp rather than a size comparison:

@Join
public class LeastActiveLoadBalance extends AbstractLoadBalancer {

    private final int recyclePeriod = 60000;

    private final ConcurrentMap<String, ActiveCount> countMap = new ConcurrentHashMap<>(16);

    private final AtomicBoolean updateLock = new AtomicBoolean();

    private volatile long lastRecycle;

    @Override
    protected Upstream doSelect(final List<Upstream> upstreamList, final LoadBalanceData data) {
        long now = System.currentTimeMillis();
        Map<String, Upstream> domainMap = upstreamList.stream()
                .collect(Collectors.toConcurrentMap(Upstream::buildDomain, upstream -> upstream));

        domainMap.keySet().forEach(domain -> {
            ActiveCount activeCount = countMap.computeIfAbsent(domain, key -> new ActiveCount(now));
            activeCount.setLastUpdate(now);
        });

        final String domain = countMap.entrySet().stream()
                .filter(entry -> domainMap.containsKey(entry.getKey()))
                .min(Comparator.comparingLong(entry -> entry.getValue().getCount()))
                .map(Map.Entry::getKey)
                .orElse(upstreamList.get(0).buildDomain());

        ActiveCount activeCount = countMap.get(domain);
        if (Objects.nonNull(activeCount)) {
            activeCount.increase();
        }

        if (!updateLock.get() && now - lastRecycle > recyclePeriod && updateLock.compareAndSet(false, true)) {
            try {
                countMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > recyclePeriod);
                lastRecycle = now;
            } finally {
                updateLock.set(false);
            }
        }
        return domainMap.get(domain);
    }

    protected static class ActiveCount {

        private final AtomicLong count = new AtomicLong(Long.MIN_VALUE);

        private volatile long lastUpdate;

        ActiveCount(final long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }

        void increase() {
            count.addAndGet(1);
        }

        long getCount() {
            return count.get();
        }

        long getLastUpdate() {
            return lastUpdate;
        }

        void setLastUpdate(final long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }
}

Key points:

• Live entries are refreshed on every request, so they are never stale and can't be evicted by
  another selector's cleanup — no cross-selector deletion.
• Only entries whose lastUpdate is older than RECYCLE_PERIOD are evicted, i.e. exactly the
  upstreams that were removed from the lists — the leak is bounded to ~RECYCLE_PERIOD.
• Eviction is throttled to at most once per RECYCLE_PERIOD and guarded by an update lock, so steady
  state has no per-request cleanup overhead and no concurrent eviction.

I'll update the tests to cover stale-entry eviction and the multi-selector isolation case. Happy to
adjust RECYCLE_PERIOD or the eviction gating if you'd prefer a different approach — please let me
know if you see any issues.

[nikhiln64]

This looks like the right direction, and you are right that the size gate does not work here. On a single shared map countMap.size() is the union of every selector's domains, so it is not comparable to one selector's upstreamList.size(), and it cannot see a same length member swap, so my suggestion would have left the cross selector deletion in place. The time based recycle avoids that cleanly. Refreshing lastUpdate for every live domain on each request means an entry can only be evicted after it has genuinely gone quiet for recyclePeriod, so one selector's cleanup can no longer drop another selector's live entries and the leak is bounded to about recyclePeriod. The updateLock and the once per period throttle keep the cleanup off the hot path, and the selected domain is always one you refreshed this call, so the countMap.get(domain) with the Objects.nonNull guard cannot be caught by a concurrent removeIf. The concurrency reads as sound to me.

The one thing I would think about is the Long.MIN_VALUE seed on a revived entry. If an upstream goes quiet longer than recyclePeriod it gets recycled, and when traffic returns its ActiveCount is recreated at Long.MIN_VALUE. Because selection picks the minimum count, that upstream is then chosen on almost every request until its counter climbs back to its peers, and since the count is cumulative and never decremented that catch up can take as many requests as the peers have served in total, so a node that just came back can absorb nearly all the traffic for a long stretch rather than a brief burst. A brand new upstream already has this property today, but recycling means an idle then active upstream hits it too. It might be worth seeding a new or revived entry to the current minimum of its live peers rather than Long.MIN_VALUE, so it rejoins at parity. Not a blocker, more a behavior to decide on consciously.

On the tests, the stale eviction and multi selector isolation cases you mentioned are the ones I would want to see. One more worth adding is the revive case: let an entry age past recyclePeriod so it is recycled, then send traffic again and assert the distribution does not collapse onto the revived node. Thanks for turning this around so quickly.