PIP-45: Switch ManagedLedger to use MetadataStore interface

managed-ledger/pom.xml

@@ -60,6 +60,12 @@
       <artifactId>pulsar-common</artifactId>
       <version>${project.version}</version>
     </dependency>
+
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-metadata</artifactId>
+      <version>${project.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>com.google.guava</groupId>

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

@@ -39,21 +39,33 @@ public static ManagedLedgerException getManagedLedgerException(Throwable e) {
     }
 
     public static class MetaStoreException extends ManagedLedgerException {
-        public MetaStoreException(Exception e) {
-            super(e);
+        public MetaStoreException(Throwable t) {
+            super(t);
+        }
+
+        public MetaStoreException(String msg) {
+            super(msg);
         }
     }
 
     public static class BadVersionException extends MetaStoreException {
         public BadVersionException(Exception e) {
             super(e);
         }
+
+        public BadVersionException(String msg) {
+            super(msg);
+        }
     }
 
     public static class MetadataNotFoundException extends MetaStoreException {
         public MetadataNotFoundException(Exception e) {
             super(e);
         }
+
+        public MetadataNotFoundException(String msg) {
+            super(msg);
+        }
     }
 
 

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

@@ -24,7 +24,7 @@
 @SuppressWarnings("checkstyle:javadoctype")
 public class ManagedLedgerInfo {
     /** Z-Node version. */
-    public int version;
+    public long version;
     public String creationDate;
     public String modificationDate;
 
@@ -42,7 +42,7 @@ public static class LedgerInfo {
 
     public static class CursorInfo {
         /** Z-Node version. */
-        public int version;
+        public long version;
         public String creationDate;
         public String modificationDate;
 

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

@@ -83,7 +83,6 @@
 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.impl.MetaStore.Stat;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
@@ -93,6 +92,7 @@
 import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
+import org.apache.pulsar.metadata.api.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

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

@@ -68,15 +68,16 @@
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.ManagedLedgerInitializeLedgerCallback;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.State;
 import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
-import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange;
 import org.apache.bookkeeper.mledger.util.Futures;
 import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
 import org.apache.pulsar.common.util.DateFormatter;
-import org.apache.zookeeper.KeeperException;
+import org.apache.pulsar.metadata.api.MetadataStore;
+import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.impl.zookeeper.ZKMetadataStore;
 import org.apache.zookeeper.ZooKeeper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -101,6 +102,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
     private final ScheduledFuture<?> statsTask;
 
     private final long cacheEvictionTimeThresholdNanos;
+    private final MetadataStore metadataStore;
 
     private static final int StatsPeriodSeconds = 60;
 
@@ -160,7 +162,8 @@ private ManagedLedgerFactoryImpl(BookkeeperFactoryForCustomEnsemblePlacementPoli
         this.bookkeeperFactory = bookKeeperGroupFactory;
         this.isBookkeeperManaged = isBookkeeperManaged;
         this.zookeeper = isBookkeeperManaged ? zooKeeper : null;
-        this.store = new MetaStoreImplZookeeper(zooKeeper, orderedExecutor);
+        this.metadataStore = new ZKMetadataStore(zooKeeper);
+        this.store = new MetaStoreImpl(metadataStore, orderedExecutor);
         this.config = config;
         this.mbean = new ManagedLedgerFactoryMBeanImpl(this);
         this.entryCacheManager = new EntryCacheManager(this);
@@ -464,6 +467,11 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
         cacheEvictionExecutor.shutdownNow();
 
         entryCacheManager.clear();
+        try {
+            metadataStore.close();
+        } catch (Exception e) {
+            throw new ManagedLedgerException(e);
+        }
     }
 
     @Override

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

@@ -106,7 +106,6 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
 import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
-import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
 import org.apache.bookkeeper.mledger.offload.OffloadUtils;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
@@ -117,6 +116,7 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition;
 import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
+import org.apache.pulsar.metadata.api.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

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

@@ -41,6 +41,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.PersistentOfflineTopicStats;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.metadata.api.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -146,7 +147,7 @@ private void readLedgerMeta(final ManagedLedgerFactoryImpl factory, final TopicN
         store.getManagedLedgerInfo(managedLedgerName, false /* createIfMissing */,
                 new MetaStore.MetaStoreCallback<MLDataFormats.ManagedLedgerInfo>() {
                     @Override
-                    public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, MetaStore.Stat version) {
+                    public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, Stat stat) {
                         for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ls : mlInfo.getLedgerInfoList()) {
                             ledgers.put(ls.getLedgerId(), ls);
                         }
@@ -230,7 +231,7 @@ private void calculateCursorBacklogs(final ManagedLedgerFactoryImpl factory, fin
 
         store.getCursors(managedLedgerName, new MetaStore.MetaStoreCallback<List<String>>() {
             @Override
-            public void operationComplete(List<String> cursors, MetaStore.Stat v) {
+            public void operationComplete(List<String> cursors, Stat v) {
                 // Load existing cursors
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] Found {} cursors", managedLedgerName, cursors.size());
@@ -336,7 +337,7 @@ public void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> seq,
                             new MetaStore.MetaStoreCallback<MLDataFormats.ManagedCursorInfo>() {
                                 @Override
                                 public void operationComplete(MLDataFormats.ManagedCursorInfo info,
-                                        MetaStore.Stat version) {
+                                        Stat stat) {
                                     long cursorLedgerId = info.getCursorsLedgerId();
                                     if (log.isDebugEnabled()) {
                                         log.debug("[{}] Cursor {} meta-data read ledger id {}", managedLedgerName,

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

@@ -22,19 +22,13 @@
 import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
+import org.apache.pulsar.metadata.api.Stat;
 
 /**
  * Interface that describes the operations that the ManagedLedger need to do on the metadata store.
  */
 public interface MetaStore {
 
-    @SuppressWarnings("checkstyle:javadoctype")
-    interface Stat {
-        int getVersion();
-        long getCreationTimestamp();
-        long getModificationTimestamp();
-    }
-
     @SuppressWarnings("checkstyle:javadoctype")
     interface UpdateLedgersIdsCallback {
         void updateLedgersIdsComplete(MetaStoreException status, Stat stat);
@@ -64,12 +58,12 @@ interface MetaStoreCallback<T> {
      *            the name of the ManagedLedger
      * @param mlInfo
      *            managed ledger info
-     * @param version
+     * @param stat
      *            version object associated with current state
      * @param callback
      *            callback object
      */
-    void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat version,
+    void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat stat,
             MetaStoreCallback<Void> callback);
 
     /**
@@ -98,12 +92,12 @@ void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat vers
      *            the name of the ManagedLedger
      * @param cursorName
      * @param info
-     * @param version
+     * @param stat
      * @param callback
      *            the callback
      * @throws MetaStoreException
      */
-    void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Stat version,
+    void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Stat stat,
             MetaStoreCallback<Void> callback);
 
     /**