PIP 81: Split the individual acknowledgments into multiple entries

conf/broker.conf

@@ -950,6 +950,13 @@ managedLedgerMaxUnackedRangesToPersist=10000
 # zookeeper.
 managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000
 
+# If enabled, the maximum "acknowledgment holes" will not be limited.
+# Only part of hotspot data will keep in memory by using LRU.
+enableLruCacheMaxUnackedRanges=false
+
+# Max Bytes of UnackedRanges can keep in memory.
+maxUnackedRangesInMemoryBytes=3145728
+
 # Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
 # corrupted at bookkeeper and managed-cursor is stuck at that ledger.
 autoSkipNonRecoverableData=false

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

@@ -46,6 +46,8 @@ public class ManagedLedgerConfig {
     private boolean createIfMissing = true;
     private int maxUnackedRangesToPersist = 10000;
     private int maxBatchDeletedIndexToPersist = 10000;
+    private boolean enableLruCacheMaxUnackedRanges = false;
+    private long maxUnackedRangesInMemoryBytes = 3145728;
     private boolean deletionAtBatchIndexLevelEnabled = true;
     private int maxUnackedRangesToPersistInZk = 1000;
     private int maxEntriesPerLedger = 50000;
@@ -488,6 +490,24 @@ public void setMaxUnackedRangesToPersistInZk(int maxUnackedRangesToPersistInZk)
         this.maxUnackedRangesToPersistInZk = maxUnackedRangesToPersistInZk;
     }
 
+    public boolean isEnableLruCacheMaxUnackedRanges() {
+        return enableLruCacheMaxUnackedRanges;
+    }
+
+    public ManagedLedgerConfig setEnableLruCacheMaxUnackedRanges(boolean enableLruCacheMaxUnackedRanges) {
+        this.enableLruCacheMaxUnackedRanges = enableLruCacheMaxUnackedRanges;
+        return this;
+    }
+
+    public long getMaxUnackedRangesInMemoryBytes() {
+        return maxUnackedRangesInMemoryBytes;
+    }
+
+    public ManagedLedgerConfig setMaxUnackedRangesInMemoryBytes(long maxUnackedRangesInMemoryBytes) {
+        this.maxUnackedRangesInMemoryBytes = maxUnackedRangesInMemoryBytes;
+        return this;
+    }
+
     /**
      * Get ledger offloader which will be used to offload ledgers to longterm storage.
      *