[WIP] Add cacheEvictionByExpectedReadCount solution

.github/workflows/pulsar-ci-flaky.yaml

@@ -24,6 +24,7 @@ on:
       - master
       - branch-*
       - pulsar-*
+      - lh-*
   schedule:
     # scheduled job with JDK 21
     - cron: '0 12 * * *'

.github/workflows/pulsar-ci.yaml

@@ -24,6 +24,7 @@ on:
       - master
       - branch-*
       - pulsar-*
+      - lh-*
   schedule:
     # scheduled job with JDK 21
     - cron: '0 12 * * *'

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

@@ -66,4 +66,6 @@ public interface Entry {
      * of data reached to 0).
      */
     boolean release();
+
+    EntryReadCountHandler getReadCountHandler();
 }

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

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger;
+
+public interface EntryReadCountHandler {
+    int getExpectedReadCount();
+    boolean incrementExpectedReadCount();
+    boolean incrementExpectedReadCount(int increment);
+    void markRead();
+    void markEvicted();
+}

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

@@ -83,6 +83,9 @@ public class ManagedLedgerConfig {
     @Getter
     @Setter
     private boolean cacheEvictionByMarkDeletedPosition = false;
+    @Getter
+    @Setter
+    private boolean cacheEvictionByExpectedReadCount = true;
     private int minimumBacklogCursorsForCaching = 0;
     private int minimumBacklogEntriesForCaching = 1000;
     private int maxBacklogBetweenCursorsForCaching = 1000;

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

@@ -22,6 +22,7 @@
 import io.netty.util.Recycler;
 import io.netty.util.ReferenceCounted;
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.EntryReadCountHandler;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionFactory;
 import org.apache.bookkeeper.mledger.util.AbstractCASReferenceCounted;
@@ -36,6 +37,7 @@ public abstract class AbstractEntryImpl<T extends AbstractEntryImpl<T>> extends
     private int length = -1;
     private Position position;
     private Runnable onDeallocate;
+    protected EntryReadCountHandlerImpl readCountHandler;
 
     public AbstractEntryImpl(Recycler.Handle<T> recyclerHandle) {
         this.recyclerHandle = recyclerHandle;
@@ -150,6 +152,7 @@ protected final void deallocate() {
         ledgerId = -1;
         entryId = -1;
         position = null;
+        readCountHandler = null;
         beforeRecycle();
         recyclerHandle.recycle(self());
     }
@@ -175,4 +178,9 @@ protected void beforeRecycle() {
     protected T self() {
         return (T) this;
     }
+
+    @Override
+    public EntryReadCountHandler getReadCountHandler() {
+        return readCountHandler;
+    }
 }

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

@@ -36,7 +36,10 @@ protected EntryImpl newObject(Handle<EntryImpl> handle) {
         }
     };
 
-    public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor) {
+    private boolean decreaseReadCountOnRelease;
+
+    public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor,
+                                   int expectedReadCount) {
         ManagedLedgerInterceptor.PayloadProcessorHandle processorHandle = null;
         if (interceptor != null) {
             ByteBuf duplicateBuffer = ledgerEntry.getEntryBuffer().retainedDuplicate();
@@ -49,20 +52,21 @@ public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor
                 duplicateBuffer.release();
             }
         }
-        EntryImpl returnEntry = create(ledgerEntry);
+        EntryImpl returnEntry = create(ledgerEntry, expectedReadCount);
         if (processorHandle != null) {
             processorHandle.release();
             ledgerEntry.close();
         }
         return returnEntry;
     }
 
-    public static EntryImpl create(LedgerEntry ledgerEntry) {
+    public static EntryImpl create(LedgerEntry ledgerEntry, int expectedReadCount) {
         EntryImpl entry = RECYCLER.get();
         entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerEntry.getLedgerId();
         entry.entryId = ledgerEntry.getEntryId();
         entry.setDataBuffer(ledgerEntry.getEntryBuffer().retain());
+        entry.readCountHandler = EntryReadCountHandlerImpl.create(expectedReadCount);
         entry.setRefCnt(1);
         return entry;
     }
@@ -79,11 +83,18 @@ public static EntryImpl create(long ledgerId, long entryId, byte[] data) {
     }
 
     public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) {
+        return create(ledgerId, entryId, data, 0);
+    }
+
+    public static EntryImpl create(long ledgerId, long entryId, ByteBuf data, int expectedReadCount) {
         EntryImpl entry = RECYCLER.get();
         entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerId;
         entry.entryId = entryId;
         entry.setDataBuffer(data.retain());
+        if (expectedReadCount > 0) {
+            entry.readCountHandler = EntryReadCountHandlerImpl.create(expectedReadCount);
+        }
         entry.setRefCnt(1);
         return entry;
     }
@@ -103,12 +114,32 @@ public static EntryImpl create(Entry other) {
         entry.timestamp = System.nanoTime();
         entry.ledgerId = other.getLedgerId();
         entry.entryId = other.getEntryId();
+        entry.readCountHandler = (EntryReadCountHandlerImpl) other.getReadCountHandler();
         entry.setDataBuffer(other.getDataBuffer().retainedDuplicate());
         entry.setRefCnt(1);
+        entry.decreaseReadCountOnRelease = true;
         return entry;
     }
 
     private EntryImpl(Recycler.Handle<EntryImpl> recyclerHandle) {
         super(recyclerHandle);
     }
+
+    public void setDecreaseReadCountOnRelease(boolean enabled) {
+        decreaseReadCountOnRelease = enabled;
+    }
+
+    @Override
+    protected void beforeDeallocate() {
+        super.beforeDeallocate();
+        if (decreaseReadCountOnRelease && readCountHandler != null) {
+            readCountHandler.markRead();
+        }
+    }
+
+    @Override
+    protected void beforeRecycle() {
+        super.beforeRecycle();
+        decreaseReadCountOnRelease = false;
+    }
 }

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

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import org.apache.bookkeeper.mledger.EntryReadCountHandler;
+
+public class EntryReadCountHandlerImpl implements EntryReadCountHandler {
+    private static final int EVICTED_VALUE = Integer.MIN_VALUE / 2;
+    private static AtomicIntegerFieldUpdater<EntryReadCountHandlerImpl> expectedReadCountUpdater =
+            AtomicIntegerFieldUpdater.newUpdater(EntryReadCountHandlerImpl.class, "expectedReadCount");
+
+    private volatile int expectedReadCount;
+
+    private EntryReadCountHandlerImpl(int expectedReadCount) {
+        this.expectedReadCount = expectedReadCount;
+    }
+
+    public int getExpectedReadCount() {
+        return expectedReadCount;
+    }
+
+    @Override
+    public boolean incrementExpectedReadCount(int increment) {
+        int newValue = expectedReadCountUpdater.updateAndGet(this, current -> {
+            // If the value is EVICTED_VALUE, we don't allow incrementing it anymore
+            if (current == EVICTED_VALUE) {
+                return current;
+            } else {
+                return current + increment;
+            }
+        });
+        return newValue != EVICTED_VALUE;
+    }
+
+    @Override
+    public boolean incrementExpectedReadCount() {
+        return incrementExpectedReadCount(1);
+    }
+
+    @Override
+    public void markRead() {
+        expectedReadCountUpdater.updateAndGet(this, current -> {
+            // If the value is EVICTED_VALUE, don't change it
+            if (current == EVICTED_VALUE) {
+                return current;
+            } else {
+                return current - 1;
+            }
+        });
+    }
+
+    @Override
+    public void markEvicted() {
+        expectedReadCountUpdater.set(this, EVICTED_VALUE);
+    }
+
+    public static EntryReadCountHandlerImpl create(int expectedReadCount) {
+        return new EntryReadCountHandlerImpl(expectedReadCount);
+    }
+}