[WIP]Evict usage of ManagedLedgerImpl in ManagedLedgerFactoryImpl

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

@@ -20,11 +20,9 @@
 
 import com.google.common.base.Predicate;
 import com.google.common.collect.Range;
-
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
 import org.apache.bookkeeper.common.annotation.InterfaceStability;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
@@ -234,6 +232,8 @@ void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntriesCallba
      */
     boolean hasMoreEntries();
 
+    boolean hasMoreEntries(PositionImpl position);
+
     /**
      * Return the number of messages that this cursor still has to read.
      *

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

@@ -20,6 +20,7 @@
 
 import io.netty.buffer.ByteBuf;
 import java.util.Map;
+import java.util.NavigableMap;
 import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
 import org.apache.bookkeeper.common.annotation.InterfaceStability;
@@ -30,6 +31,8 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.OffloadCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenCursorCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.TerminateCallback;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
 import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
@@ -59,6 +62,19 @@
 @InterfaceStability.Stable
 public interface ManagedLedger {
 
+    /**
+     * Make ManagedLedger ready to work
+     * @param callback
+     * @param ctx
+     */
+    void initialize(final ManagedLedgerInitializeLedgerCallback callback, final Object ctx);
+
+    boolean isValidPosition(PositionImpl nextReadPosition);
+
+    boolean hasMoreEntries(PositionImpl nextReadPosition);
+
+    void addWaitingEntryCallBack(WaitingEntryCallBack streamingEntryReader);
+
     /**
      * @return the unique name of this ManagedLedger
      */
@@ -100,6 +116,8 @@ public interface ManagedLedger {
      */
     void asyncAddEntry(byte[] data, AddEntryCallback callback, Object ctx);
 
+    void asyncReadEntry(PositionImpl position, AsyncCallbacks.ReadEntryCallback callback, Object ctx);
+
     /**
      * Append a new entry to the end of a managed ledger.
      *
@@ -357,6 +375,26 @@ public interface ManagedLedger {
      */
     long getNumberOfEntries();
 
+    long getEntriesAddedCounter();
+
+    long getLastLedgerCreatedTimestamp();
+
+    long getLastLedgerCreationFailureTimestamp();
+
+    int getWaitingCursorsCount();
+
+    long getCurrentLedgerEntries();
+
+    long getCurrentLedgerSize();
+
+    NavigableMap<Long, LedgerInfo> getLedgersInfo();
+
+    CompletableFuture<String> getLedgerMetadata(long ledgerId);
+
+    boolean ledgerExists(long ledgerId);
+
+    void asyncDeleteLedgerFromBookKeeper(long ledgerId);
+
     /**
      * Get the total number of active entries for this managed ledger.
      *
@@ -387,6 +425,23 @@ public interface ManagedLedger {
      */
     long getEstimatedBacklogSize();
 
+    /**
+     * Get estimated backlog size from a specific position.
+     * @param pos
+     * @return
+     */
+    long getEstimatedBacklogSize(PositionImpl pos);
+
+    /**
+     * number of entries are in add progress
+     */
+    int getPendingAddEntriesCount();
+
+    /**
+     * Get the total size in bytes of all the entries stored in this cache.
+     */
+    long getCacheSize();
+
     /**
      * Return the size of all ledgers offloaded to 2nd tier storage
      */
@@ -430,6 +485,13 @@ public interface ManagedLedger {
      */
     ManagedLedgerMXBean getStats();
 
+    /**
+     * Remove all entries already read by active cursors and
+     * remove entries older than the cutoff threshold
+     * @param maxTimestamp
+     */
+    void doCacheEviction(long maxTimestamp);
+
     /**
      * Delete the ManagedLedger.
      *
@@ -504,6 +566,12 @@ public interface ManagedLedger {
      */
     Position getLastConfirmedEntry();
 
+    /**
+     * Get state of the managed ledger.
+     * @return
+     */
+    String getState();
+
     /**
      * Signaling managed ledger that we can resume after BK write failure
      */
@@ -590,6 +658,27 @@ void asyncSetProperties(Map<String, String> properties, final AsyncCallbacks.Upd
      * */
     CompletableFuture<Position> asyncFindPosition(com.google.common.base.Predicate<Entry> predicate);
 
+    /**
+     * Get the entry position at a given distance from a given position.
+     *
+     * @param startPosition
+     *            starting position
+     * @param n
+     *            number of entries to skip ahead
+     * @param startRange
+     *            specifies whether or not to include the start position in calculating the distance
+     * @return the new position that is n entries ahead
+     */
+    PositionImpl getPositionAfterN(final PositionImpl startPosition, long n,
+                                   ManagedLedgerImpl.PositionBound startRange);
+
+    /**
+     * first position of current topic
+     */
+    PositionImpl getFirstPosition();
+
+    PositionImpl getLastPosition();
+
     /**
      * Get the ManagedLedgerInterceptor for ManagedLedger.
      * */
@@ -600,4 +689,21 @@ void asyncSetProperties(Map<String, String> properties, final AsyncCallbacks.Upd
      * will got null if corresponding ledger not exists.
      */
     CompletableFuture<LedgerInfo> getLedgerInfo(long ledgerId);
+
+    /**
+     * get the valid position next to the given one
+     * @param position current postion
+     */
+    PositionImpl getNextValidPosition(final PositionImpl position);
+
+    interface ManagedLedgerInitializeLedgerCallback {
+        void initializeComplete();
+
+        void initializeFailed(ManagedLedgerException e);
+    }
+
+    // define boundaries for position based seeks and searches
+    enum PositionBound {
+        startIncluded, startExcluded
+    }
 }

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

@@ -19,7 +19,6 @@
 package org.apache.bookkeeper.mledger;
 
 import java.util.function.Supplier;
-
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
 import org.apache.bookkeeper.common.annotation.InterfaceStability;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteLedgerCallback;
@@ -113,7 +112,9 @@ ReadOnlyCursor openReadOnlyCursor(String managedLedgerName, Position startPositi
      * @param ctx
      */
     void asyncOpenReadOnlyCursor(String managedLedgerName, Position startPosition, ManagedLedgerConfig config,
-            OpenReadOnlyCursorCallback callback, Object ctx);
+                                 OpenReadOnlyCursorCallback callback, Object ctx);
+
+    void close(ManagedLedger ledger);
 
     /**
      * Get the current metadata info for a managed ledger.

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

@@ -26,7 +26,6 @@
 import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException;
 import static org.apache.bookkeeper.mledger.util.Errors.isNoSuchLedgerExistsException;
 import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
-
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Predicate;
@@ -38,9 +37,7 @@
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.RateLimiter;
 import com.google.protobuf.InvalidProtocolBufferException;
-
 import io.netty.util.concurrent.FastThreadLocal;
-
 import java.time.Clock;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -62,7 +59,6 @@
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
-
 import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
 import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
 import org.apache.bookkeeper.client.AsyncCallback.OpenCallback;
@@ -80,15 +76,15 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedCursorMXBean;
 import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedger.PositionBound;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.CursorAlreadyClosedException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.NoMoreEntriesToReadException;
-import org.apache.bookkeeper.mledger.ManagedCursorMXBean;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound;
 import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty;
@@ -584,6 +580,26 @@ public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadE
         ledger.asyncReadEntries(op);
     }
 
+    public void asyncReadEntries(int restNumOfEntries, long restMaxSizeBytes, ReadEntriesCallback callback,
+                                 Object ctx, PositionImpl maxPosition, List<Entry> alreadyRead) {
+        checkArgument(restNumOfEntries >= 0);
+        if (isClosed()) {
+            callback.readEntriesFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+            return;
+        }
+
+        int restNumOfEntriesToRead = applyMaxSizeCap(restNumOfEntries, restMaxSizeBytes);
+
+        PENDING_READ_OPS_UPDATER.incrementAndGet(this);
+        OpReadEntry op = OpReadEntry
+                .create(this, readPosition, restNumOfEntriesToRead + alreadyRead.size(), callback, ctx, maxPosition);
+        if (!alreadyRead.isEmpty()) {
+            op.readEntriesComplete(alreadyRead, ctx);
+        } else {
+            ledger.asyncReadEntries(op);
+        }
+    }
+
     @Override
     public Entry getNthEntry(int n, IndividualDeletedEntries deletedEntries)
             throws InterruptedException, ManagedLedgerException {
@@ -1191,17 +1207,7 @@ public Set<? extends Position> asyncReplayEntries(Set<? extends Position> positi
         }
 
         // filters out messages which are already acknowledged
-        Set<Position> alreadyAcknowledgedPositions = Sets.newHashSet();
-        lock.readLock().lock();
-        try {
-            positions.stream()
-                    .filter(position -> individualDeletedMessages.contains(((PositionImpl) position).getLedgerId(),
-                            ((PositionImpl) position).getEntryId())
-                            || ((PositionImpl) position).compareTo(markDeletePosition) <= 0)
-                    .forEach(alreadyAcknowledgedPositions::add);
-        } finally {
-            lock.readLock().unlock();
-        }
+        Set<Position> alreadyAcknowledgedPositions = filterAlreadyAcknowledgedCallback(positions);
 
         final int totalValidPositions = positions.size() - alreadyAcknowledgedPositions.size();
         final AtomicReference<ManagedLedgerException> exception = new AtomicReference<>();
@@ -1256,6 +1262,21 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx)
         return alreadyAcknowledgedPositions;
     }
 
+    protected Set<Position> filterAlreadyAcknowledgedCallback(Set<? extends Position> positions) {
+        Set<Position> alreadyAcknowledgedPositions = Sets.newHashSet();
+        lock.readLock().lock();
+        try {
+            positions.stream()
+                    .filter(position -> individualDeletedMessages.contains(position.getLedgerId(),
+                            position.getEntryId())
+                            || ((PositionImpl) position).compareTo(markDeletePosition) <= 0)
+                    .forEach(alreadyAcknowledgedPositions::add);
+        } finally {
+            lock.readLock().unlock();
+        }
+        return alreadyAcknowledgedPositions;
+    }
+
     protected long getNumberOfEntries(Range<PositionImpl> range) {
         long allEntries = ledger.getNumberOfEntries(range);
 
@@ -1494,7 +1515,7 @@ long getNumIndividualDeletedEntriesToSkip(long numEntries) {
         return tempDeletedMessages.get();
     }
 
-    boolean hasMoreEntries(PositionImpl position) {
+    public boolean hasMoreEntries(PositionImpl position) {
         PositionImpl lastPositionInLedger = ledger.getLastPosition();
         if (position.compareTo(lastPositionInLedger) <= 0) {
             return getNumberOfEntries(Range.closed(position, lastPositionInLedger)) > 0;