Persist and recover individual deleted messages

conf/broker.conf

@@ -198,6 +198,8 @@ managedLedgerCursorMaxEntriesPerLedger=50000
 # Max time before triggering a rollover on a cursor ledger
 managedLedgerCursorRolloverTimeInSeconds=14400
 
+managedLedgerMaxUnackedRangesToPersist=1000
+
 
 
 ### --- Load balancer --- ###

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

@@ -15,9 +15,7 @@
  */
 package org.apache.bookkeeper.mledger;
 
-import java.util.List;
-import java.util.Set;
-
+import com.google.common.annotations.Beta;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
@@ -26,8 +24,10 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback;
 
-import com.google.common.annotations.Beta;
-import com.google.common.base.Predicate;
+
+import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
 
 /**
  * A ManangedCursor is a persisted cursor inside a ManagedLedger.
@@ -326,7 +326,7 @@ public void asyncSkipEntries(int numEntriesToSkip, IndividualDeletedEntries dele
      *            opaque context
      */
     public void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate<Entry> condition,
-            FindEntryCallback callback, Object ctx);
+                                        FindEntryCallback callback, Object ctx);
 
     /**
      * reset the cursor to specified position to enable replay of messages

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

@@ -31,6 +31,7 @@
 @Beta
 public class ManagedLedgerConfig {
 
+    private int maxUnackedRangesToPersist = 1000;
     private int maxEntriesPerLedger = 50000;
     private int maxSizePerLedgerMb = 100;
     private int minimumRolloverTimeMs = 0;
@@ -347,4 +348,21 @@ public ManagedLedgerConfig setRetentionSizeInMB(long retentionSizeInMB) {
     public long getRetentionSizeInMB() {
         return retentionSizeInMB;
     }
+
+    /**
+     * @return max unacked message ranges that will be persisted and recovered.
+     *
+     */
+    public int getMaxUnackedRangesToPersist() {
+        return maxUnackedRangesToPersist;
+    }
+
+    /**
+     * @param maxUnackedRangesToPersist
+     *            max unacked message ranges that will be persisted and receverd.
+     */
+    public ManagedLedgerConfig setMaxUnackedRangesToPersist(int maxUnackedRangesToPersist) {
+        this.maxUnackedRangesToPersist = maxUnackedRangesToPersist;
+        return this;
+    }
 }

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

@@ -15,13 +15,14 @@
  */
 package org.apache.bookkeeper.mledger.impl;
 
-import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound;
-import com.google.common.base.Predicate;
+
+import java.util.function.Predicate;
 
 /**
  */
@@ -43,7 +44,7 @@ enum State {
     State state;
 
     public OpFindNewest(ManagedCursorImpl cursor, PositionImpl startPosition, Predicate<Entry> condition,
-            long numberOfEntries, FindEntryCallback callback, Object ctx) {
+                        long numberOfEntries, FindEntryCallback callback, Object ctx) {
         this.cursor = cursor;
         this.startPosition = startPosition;
         this.callback = callback;
@@ -61,7 +62,7 @@ public OpFindNewest(ManagedCursorImpl cursor, PositionImpl startPosition, Predic
     public void readEntryComplete(Entry entry, Object ctx) {
         switch (state) {
         case checkFirst:
-            if (!condition.apply(entry)) {
+            if (!condition.test(entry)) {
                 callback.findEntryComplete(null, OpFindNewest.this.ctx);
                 return;
             } else {
@@ -74,7 +75,7 @@ public void readEntryComplete(Entry entry, Object ctx) {
             }
             break;
         case checkLast:
-            if (condition.apply(entry)) {
+            if (condition.test(entry)) {
                 callback.findEntryComplete(entry.getPosition(), OpFindNewest.this.ctx);
                 return;
             } else {
@@ -85,7 +86,7 @@ public void readEntryComplete(Entry entry, Object ctx) {
             }
             break;
         case searching:
-            if (condition.apply(entry)) {
+            if (condition.test(entry)) {
                 // mid - last
                 lastMatchedPosition = entry.getPosition();
                 min = mid();

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

@@ -17,8 +17,11 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.common.collect.RangeSet;
 import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo;
 
 import com.google.common.base.Objects;
 import com.google.common.collect.ComparisonChain;
@@ -39,6 +42,12 @@ public PositionImpl(PositionInfo pi) {
         this.recyclerHandle = null;
     }
 
+    public PositionImpl(NestedPositionInfo npi) {
+        this.ledgerId = npi.getLedgerId();
+        this.entryId = npi.getEntryId();
+        this.recyclerHandle = null;
+    }
+
     public PositionImpl(long ledgerId, long entryId) {
         this.ledgerId = ledgerId;
         this.entryId = entryId;