[fix][broker]Cache invalidation due to concurrent access
@@ -196,13 +196,12 @@ public class RangeCache<Key extends Comparable<Key>, Value extends ReferenceCoun
if (entry == null || timestampExtractor.getTimestamp(entry.getValue()) > maxTimestamp) {
break;
}
-
- entry = entries.pollFirstEntry();
- if (entry == null) {
+ Value value = entry.getValue();
+ boolean removeHits = entries.remove(entry.getKey(), value);
+ if (!removeHits) {
break;
}
- Value value = entry.getValue();
removedSize += weighter.getSize(value);
removedCount++;
value.release();
