Allow to configure the managed ledger cache eviction frequency

conf/broker.conf

@@ -440,6 +440,16 @@ managedLedgerCacheSizeMB=
 # Threshold to which bring down the cache level when eviction is triggered
 managedLedgerCacheEvictionWatermark=0.9
 
+# Configure the cache eviction frequency for the managed ledger cache (evictions/sec)
+managedLedgerCacheEvictionFrequency=100.0
+
+# All entries that have stayed in cache for more than the configured time, will be evicted
+managedLedgerCacheEvictionTimeThresholdMillis=1000
+
+# Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged'
+# and thus should be set as inactive.
+managedLedgerCursorBackloggedThreshold=1000
+
 # Rate limit the amount of writes per second generated by consumer acking the messages
 managedLedgerDefaultMarkDeleteRateLimit=1.0
 

conf/standalone.conf

@@ -301,6 +301,16 @@ managedLedgerCacheSizeMB=
 # Threshold to which bring down the cache level when eviction is triggered
 managedLedgerCacheEvictionWatermark=0.9
 
+# Configure the cache eviction frequency for the managed ledger cache (evictions/sec)
+managedLedgerCacheEvictionFrequency=100.0
+
+# All entries that have stayed in cache for more than the configured time, will be evicted
+managedLedgerCacheEvictionTimeThresholdMillis=1000
+
+# Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged'
+# and thus should be set as inactive.
+managedLedgerCursorBackloggedThreshold=1000
+
 # Rate limit the amount of writes generated by consumer acking the messages
 managedLedgerDefaultMarkDeleteRateLimit=0.1
 

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

@@ -515,9 +515,9 @@ public ManagedLedgerConfig setClock(Clock clock) {
     }
 
     /**
-     * 
+     *
      * Ledger-Op (Create/Delete) timeout
-     * 
+     *
      * @return
      */
     public long getMetadataOperationsTimeoutSeconds() {
@@ -526,17 +526,17 @@ public long getMetadataOperationsTimeoutSeconds() {
 
     /**
      * Ledger-Op (Create/Delete) timeout after which callback will be completed with failure
-     * 
+     *
      * @param metadataOperationsTimeoutSeconds
      */
     public ManagedLedgerConfig setMetadataOperationsTimeoutSeconds(long metadataOperationsTimeoutSeconds) {
         this.metadataOperationsTimeoutSeconds = metadataOperationsTimeoutSeconds;
         return this;
     }
-    
+
     /**
      * Ledger read-entry timeout
-     * 
+     *
      * @return
      */
     public long getReadEntryTimeoutSeconds() {
@@ -546,22 +546,22 @@ public long getReadEntryTimeoutSeconds() {
     /**
      * Ledger read entry timeout after which callback will be completed with failure. (disable timeout by setting
      * readTimeoutSeconds <= 0)
-     * 
+     *
      * @param readTimeoutSeconds
      * @return
      */
     public ManagedLedgerConfig setReadEntryTimeoutSeconds(long readEntryTimeoutSeconds) {
         this.readEntryTimeoutSeconds = readEntryTimeoutSeconds;
         return this;
     }
-    
+
     public long getAddEntryTimeoutSeconds() {
         return addEntryTimeoutSeconds;
     }
 
     /**
      * Add-entry timeout after which add-entry callback will be failed if add-entry is not succeeded.
-     * 
+     *
      * @param addEntryTimeoutSeconds
      */
     public ManagedLedgerConfig setAddEntryTimeoutSeconds(long addEntryTimeoutSeconds) {

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

@@ -37,7 +37,18 @@ public class ManagedLedgerFactoryConfig {
     private int numManagedLedgerWorkerThreads = Runtime.getRuntime().availableProcessors();
     private int numManagedLedgerSchedulerThreads = Runtime.getRuntime().availableProcessors();
 
-    public long getMaxCacheSize() {
-        return maxCacheSize;
-    }
+    /**
+     * Frequency of cache eviction triggering. Default is 100 times per second.
+     */
+    private double cacheEvictionFrequency = 100;
+
+    /**
+     * All entries that have stayed in cache for more than the configured time, will be evicted
+     */
+    private long cacheEvictionTimeThresholdMillis = 1000;
+
+    /**
+     * Threshould to consider a cursor as "backlogged"
+     */
+    private long thresholdBackloggedCursor = 1000;
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java

@@ -50,10 +50,12 @@ public interface EntryCache extends Comparable<EntryCache> {
      * Remove from cache all the entries related to a ledger up to lastPosition included.
      *
      * @param lastPosition
-     *            the position of the last entry to be invalidated (inclusive)
+     *            the position of the last entry to be invalidated (non-inclusive)
      */
     void invalidateEntries(PositionImpl lastPosition);
 
+    void invalidateEntriesBeforeTimestamp(long timestamp);
+
     /**
      * Remove from the cache all the entries belonging to a specific ledger.
      *

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java

@@ -21,23 +21,24 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException;
-import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
 
 import com.google.common.collect.Lists;
 import com.google.common.primitives.Longs;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
+
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+
 import org.apache.bookkeeper.client.api.BKException;
 import org.apache.bookkeeper.client.api.LedgerEntry;
 import org.apache.bookkeeper.client.api.ReadHandle;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.util.RangeCache;
-import org.apache.bookkeeper.mledger.util.RangeCache.Weighter;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,12 +54,10 @@ public class EntryCacheImpl implements EntryCache {
 
     private static final double MB = 1024 * 1024;
 
-    private static final Weighter<EntryImpl> entryWeighter = EntryImpl::getLength;
-
     public EntryCacheImpl(EntryCacheManager manager, ManagedLedgerImpl ml) {
         this.manager = manager;
         this.ml = ml;
-        this.entries = new RangeCache<>(entryWeighter);
+        this.entries = new RangeCache<>(EntryImpl::getLength, EntryImpl::getTimestamp);
 
         if (log.isDebugEnabled()) {
             log.debug("[{}] Initialized managed-ledger entry cache", ml.getName());
@@ -132,7 +131,7 @@ public boolean insert(EntryImpl entry) {
     public void invalidateEntries(final PositionImpl lastPosition) {
         final PositionImpl firstPosition = PositionImpl.get(-1, 0);
 
-        Pair<Integer, Long> removed = entries.removeRange(firstPosition, lastPosition, true);
+        Pair<Integer, Long> removed = entries.removeRange(firstPosition, lastPosition, false);
         int entriesRemoved = removed.getLeft();
         long sizeRemoved = removed.getRight();
         if (log.isDebugEnabled()) {
@@ -173,7 +172,7 @@ public void asyncReadEntry(ReadHandle lh, PositionImpl position, final ReadEntry
             callback.readEntryFailed(createManagedLedgerException(t), ctx);
         }
     }
-    
+
     private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEntryCallback callback,
             final Object ctx) {
         if (log.isDebugEnabled()) {
@@ -229,7 +228,7 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole
             callback.readEntriesFailed(createManagedLedgerException(t), ctx);
         }
     }
-    
+
     @SuppressWarnings({ "unchecked", "rawtypes" })
     private void asyncReadEntry0(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader,
             final ReadEntriesCallback callback, Object ctx) {
@@ -341,5 +340,11 @@ public Pair<Integer, Long> evictEntries(long sizeToFree) {
         return evicted;
     }
 
+    @Override
+    public void invalidateEntriesBeforeTimestamp(long timestamp) {
+        long evictedSize = entries.evictLEntriesBeforeTimestamp(timestamp);
+        manager.entriesRemoved(evictedSize);
+    }
+
     private static final Logger log = LoggerFactory.getLogger(EntryCacheImpl.class);
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java

@@ -189,6 +189,10 @@ public Pair<Integer, Long> evictEntries(long sizeToFree) {
             return Pair.of(0, (long) 0);
         }
 
+        @Override
+        public void invalidateEntriesBeforeTimestamp(long timestamp) {
+        }
+
         @Override
         public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader,
                 final ReadEntriesCallback callback, Object ctx) {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

@@ -40,12 +40,14 @@ protected EntryImpl newObject(Handle<EntryImpl> handle) {
     };
 
     private final Handle<EntryImpl> recyclerHandle;
+    private long timestamp;
     private long ledgerId;
     private long entryId;
     ByteBuf data;
 
     public static EntryImpl create(LedgerEntry ledgerEntry) {
         EntryImpl entry = RECYCLER.get();
+        entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerEntry.getLedgerId();
         entry.entryId = ledgerEntry.getEntryId();
         entry.data = ledgerEntry.getEntryBuffer();
@@ -57,6 +59,7 @@ public static EntryImpl create(LedgerEntry ledgerEntry) {
     // Used just for tests
     public static EntryImpl create(long ledgerId, long entryId, byte[] data) {
         EntryImpl entry = RECYCLER.get();
+        entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerId;
         entry.entryId = entryId;
         entry.data = Unpooled.wrappedBuffer(data);
@@ -66,6 +69,7 @@ public static EntryImpl create(long ledgerId, long entryId, byte[] data) {
 
     public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) {
         EntryImpl entry = RECYCLER.get();
+        entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerId;
         entry.entryId = entryId;
         entry.data = data;
@@ -76,6 +80,7 @@ public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) {
 
     public static EntryImpl create(PositionImpl position, ByteBuf data) {
         EntryImpl entry = RECYCLER.get();
+        entry.timestamp = System.nanoTime();
         entry.ledgerId = position.getLedgerId();
         entry.entryId = position.getEntryId();
         entry.data = data;
@@ -86,6 +91,7 @@ public static EntryImpl create(PositionImpl position, ByteBuf data) {
 
     public static EntryImpl create(EntryImpl other) {
         EntryImpl entry = RECYCLER.get();
+        entry.timestamp = System.nanoTime();
         entry.ledgerId = other.ledgerId;
         entry.entryId = other.entryId;
         entry.data = other.data.retainedDuplicate();
@@ -97,6 +103,10 @@ private EntryImpl(Recycler.Handle<EntryImpl> recyclerHandle) {
         this.recyclerHandle = recyclerHandle;
     }
 
+    public long getTimestamp() {
+        return timestamp;
+    }
+
     @Override
     public ByteBuf getDataBuffer() {
         return data;
@@ -152,6 +162,7 @@ protected void deallocate() {
         // This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it
         data.release();
         data = null;
+        timestamp = -1;
         ledgerId = -1;
         entryId = -1;
         recyclerHandle.recycle(this);