[fix][ml] Fix thread safe issue with RangeCache.put and RangeCache.clear

[lhotari]

Fixes #21301
Fixes #10433

Motivation

There are 2 thread safety issues in RangeCache.

RangeCache.put:

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java

Lines 76 to 84 in 643428b

	public boolean put(Key key, Value value) {
	MutableBoolean flag = new MutableBoolean();
	entries.computeIfAbsent(key, (k) -> {
	size.addAndGet(weighter.getSize(value));
	flag.setValue(true);
	return value;
	});
	return flag.booleanValue();
	}

computeIfAbsent doesn't lock the key when ConcurrentSkipListMap is used.
You can see this in the source code of ConcurrentMap:

    default V computeIfAbsent(K key,
            Function<? super K, ? extends V> mappingFunction) {
        Objects.requireNonNull(mappingFunction);
        V oldValue, newValue;
        return ((oldValue = get(key)) == null
                && (newValue = mappingFunction.apply(key)) != null
                && (oldValue = putIfAbsent(key, newValue)) == null)
            ? newValue
            : oldValue;
    }

On the other hand, ConcurrentHashMap.computeIfAbsent does lock the key and atomically call the mappingFunction. Please check the references that @michaeljmarshall shared in the issue #21301. Kudos to @michaeljmarshall for pinpointing the issue!

Another problem is in RangeCache.clear. There should be no call to entries.clear() in that method.

Modifications

Fix the problems in RangeCache.put and RangeCache.clear.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[michaeljmarshall]

LGTM, thanks for fixing it!

[codecov-commenter]

Codecov Report

Merging #21302 (0db0464) into master (643428b) will increase coverage by 36.29%.
The diff coverage is 100.00%.

@@              Coverage Diff              @@
##             master   #21302       +/-   ##
=============================================
+ Coverage     36.91%   73.21%   +36.29%     
- Complexity    12284    32488    +20204     
=============================================
  Files          1698     1887      +189     
  Lines        130510   140196     +9686     
  Branches      14260    15437     +1177     
=============================================
+ Hits          48183   102644    +54461     
+ Misses        76016    29454    -46562     
- Partials       6311     8098     +1787     

Flag	Coverage Δ
inttests	24.16% <66.66%> (-0.13%)	⬇️
systests	24.76% <66.66%> (-0.09%)	⬇️
unittests	72.52% <100.00%> (+40.41%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...org/apache/bookkeeper/mledger/util/RangeCache.java	96.55% <100.00%> (+24.96%)	⬆️

... and 1452 files with indirect coverage changes

[codelipenghui]

Nice catch!

The new solution is also better than #15707 to avoid any issues with the entry release. In #15707, it tries to resolve the problem by getting the size before putting it into the map, because the entry release is triggered by the clear() method. The new solution can also fix the potential issue from any other cases released the entry before getting the entry size.

And the change of this line https://github.com/apache/pulsar/pull/21302/files#diff-cf9a27d8abb87f6c4fe85e6e597230e9fc102c0ca8169eed08a44218c554c5edL245 will fix the issue that the newly added items to the cache be removed without the entry release.