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

…ear (#21302)

(cherry picked from commit 70d086f)

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

@@ -27,7 +27,6 @@
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicLong;
-import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.commons.lang3.tuple.Pair;
 
 /**
@@ -74,13 +73,18 @@ public RangeCache(Weighter<Value> weighter, TimestampExtractor<Value> timestampE
      * @return whether the entry was inserted in the cache
      */
     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();
+        // retain value so that it's not released before we put it in the cache and calculate the weight
+        value.retain();
+        try {
+            if (entries.putIfAbsent(key, value) == null) {
+                size.addAndGet(weighter.getSize(value));
+                return true;
+            } else {
+                return false;
+            }
+        } finally {
+            value.release();
+        }
     }
 
     public Value get(Key key) {
@@ -238,7 +242,6 @@ public synchronized Pair<Integer, Long> clear() {
             value.release();
         }
 
-        entries.clear();
         size.getAndAdd(-removedSize);
         return Pair.of(removedCount, removedSize);
     }