[pulsar] branch branch-2.11 updated: [fix][ml] Fix thread safe issue with RangeCache.put and RangeCache.clear (#21302)

[lhotari]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.11 by this push:
     new 14a9cf7a09b [fix][ml] Fix thread safe issue with RangeCache.put and 
RangeCache.clear (#21302)
14a9cf7a09b is described below

commit 14a9cf7a09b27698631a7eb7ec12fbd819d9dd81
Author: Lari Hotari <lhot...@users.noreply.github.com>
AuthorDate: Sat Oct 7 05:26:48 2023 +0300

    [fix][ml] Fix thread safe issue with RangeCache.put and RangeCache.clear 
(#21302)
    
    (cherry picked from commit 70d086f8f35d36800059d0d68e13d0ca017bf233)
---
 .../apache/bookkeeper/mledger/util/RangeCache.java  | 21 ++++++++++++---------
 1 file changed, 12 insertions(+), 9 deletions(-)

diff --git 
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java
 
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java
index 98ae6659b78..a8d0baad72b 100644
--- 
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java
+++ 
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java
@@ -27,7 +27,6 @@ import java.util.Map;
 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 class RangeCache<Key extends Comparable<Key>, Value 
extends ReferenceCoun
      * @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 boolean exists(Key key) {
@@ -242,7 +246,6 @@ public class RangeCache<Key extends Comparable<Key>, Value 
extends ReferenceCoun
             value.release();
         }
 
-        entries.clear();
         size.getAndAdd(-removedSize);
         return Pair.of(removedCount, removedSize);
     }