BP-42: New Client API - list ledgers

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java

@@ -35,6 +35,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -58,6 +59,10 @@
 import org.apache.bookkeeper.client.api.BookKeeperBuilder;
 import org.apache.bookkeeper.client.api.CreateBuilder;
 import org.apache.bookkeeper.client.api.DeleteBuilder;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.client.api.LedgersIterator;
+import org.apache.bookkeeper.client.api.ListLedgersResult;
+import org.apache.bookkeeper.client.api.ListLedgersResultBuilder;
 import org.apache.bookkeeper.client.api.OpenBuilder;
 import org.apache.bookkeeper.client.api.WriteFlag;
 import org.apache.bookkeeper.common.allocator.ByteBufAllocatorBuilder;
@@ -71,6 +76,7 @@
 import org.apache.bookkeeper.meta.CleanupLedgerManager;
 import org.apache.bookkeeper.meta.LedgerIdGenerator;
 import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.meta.LedgerManager.LedgerRangeIterator;
 import org.apache.bookkeeper.meta.LedgerManagerFactory;
 import org.apache.bookkeeper.meta.MetadataClientDriver;
 import org.apache.bookkeeper.meta.MetadataDrivers;
@@ -86,6 +92,7 @@
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.util.EventLoopUtil;
 import org.apache.bookkeeper.util.SafeRunnable;
+import org.apache.bookkeeper.versioning.Versioned;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooKeeper;
@@ -1473,6 +1480,118 @@ public DeleteBuilder newDeleteLedgerOp() {
         return new LedgerDeleteOp.DeleteBuilderImpl(this);
     }
 
+    private static final class SyncLedgerIterator implements LedgersIterator {
+
+        private final LedgerRangeIterator iterator;
+        private final ListLedgersResultImpl parent;
+        Iterator<Long> currentRange = null;
+
+        public SyncLedgerIterator(LedgerRangeIterator iterator, ListLedgersResultImpl parent) {
+            this.iterator = iterator;
+            this.parent = parent;
+        }
+
+        @Override
+        public boolean hasNext() throws IOException {
+            parent.checkClosed();
+            if (currentRange != null) {
+                if (currentRange.hasNext()) {
+                    return true;
+                }
+            } else if (iterator.hasNext()) {
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public long next() throws IOException {
+            parent.checkClosed();
+            if (currentRange == null || !currentRange.hasNext()) {
+                currentRange = iterator.next().getLedgers().iterator();
+            }
+            return currentRange.next();
+        }
+    }
+
+    private static final class ListLedgersResultImpl implements ListLedgersResult {
+
+        private final LedgerRangeIterator iterator;
+        private boolean closed = false;
+        private LedgersIterator ledgersIterator;
+
+        public ListLedgersResultImpl(LedgerRangeIterator iterator) {
+            this.iterator = iterator;
+        }
+
+        void checkClosed() {
+            if (closed) {
+                throw new IllegalStateException("ListLedgersResult is closed");
+            }
+        }
+
+        private void initLedgersIterator() {
+            if (ledgersIterator != null) {
+                throw new IllegalStateException("LedgersIterator must be requested once");
+            }
+            ledgersIterator = new SyncLedgerIterator(iterator, this);
+        }
+
+        @Override
+        public LedgersIterator iterator() {
+            checkClosed();
+            initLedgersIterator();
+            return ledgersIterator;
+        }
+
+        @Override
+        public Iterable<Long> toIterable() {
+            checkClosed();
+            initLedgersIterator();
+
+            return () -> new Iterator<Long>() {
+                @Override
+                public boolean hasNext() {
+                    try {
+                        return ledgersIterator.hasNext();
+                    } catch (IOException ex) {
+                        throw new RuntimeException(ex);
+                    }
+                }
+
+                @Override
+                public Long next() {
+                    try {
+                        return ledgersIterator.next();
+                    } catch (IOException ex) {
+                        throw new RuntimeException(ex);
+                    }
+                }
+            };
+        }
+
+        @Override
+        public void close() throws Exception {
+            closed = true;
+        }
+    }
+
+    @Override
+    public ListLedgersResultBuilder newListLedgersOp() {
+        return () -> {
+            final LedgerRangeIterator iterator = getLedgerManager().getLedgerRanges(0);
+            return CompletableFuture.completedFuture(new ListLedgersResultImpl(iterator));
+        };
+    }
+
+    @Override
+    public CompletableFuture<LedgerMetadata> getLedgerMetadata(long ledgerId) {
+        CompletableFuture<Versioned<LedgerMetadata>> versioned = getLedgerManager().readLedgerMetadata(ledgerId);
+        return versioned.thenApply(versionedLedgerMetadata -> {
+            return versionedLedgerMetadata.getValue();
+        });
+    }
+
     private final ClientContext clientCtx = new ClientContext() {
             @Override
             public ClientInternalConf getConf() {

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java

@@ -147,17 +147,17 @@ public void initiate() {
         }
 
 
-        this.metadata = metadataBuilder.build();
         if (this.generateLedgerId) {
-            generateLedgerIdAndCreateLedger();
+            generateLedgerIdAndCreateLedger(metadataBuilder);
         } else {
+            this.metadata = metadataBuilder.withId(ledgerId).build();
             // Create ledger with supplied ledgerId
             bk.getLedgerManager().createLedgerMetadata(ledgerId, metadata)
-                .whenComplete((written, exception) -> metadataCallback(written, exception));
+                .whenComplete((written, exception) -> metadataCallback(written, exception, metadataBuilder));
         }
     }
 
-    void generateLedgerIdAndCreateLedger() {
+    void generateLedgerIdAndCreateLedger(LedgerMetadataBuilder metadataBuilder) {
         // generate a ledgerId
         final LedgerIdGenerator ledgerIdGenerator = bk.getLedgerIdGenerator();
         ledgerIdGenerator.generateLedgerId(new GenericCallback<Long>() {
@@ -168,9 +168,10 @@ public void operationComplete(int rc, Long ledgerId) {
                     return;
                 }
                 LedgerCreateOp.this.ledgerId = ledgerId;
+                LedgerCreateOp.this.metadata = metadataBuilder.withId(ledgerId).build();
                 // create a ledger with metadata
                 bk.getLedgerManager().createLedgerMetadata(ledgerId, metadata)
-                    .whenComplete((written, exception) -> metadataCallback(written, exception));
+                    .whenComplete((written, exception) -> metadataCallback(written, exception, metadataBuilder));
             }
         });
     }
@@ -190,12 +191,13 @@ public void initiateAdv(final long ledgerId) {
     /**
      * Callback when metadata store has responded.
      */
-    private void metadataCallback(Versioned<LedgerMetadata> writtenMetadata, Throwable exception) {
+    private void metadataCallback(Versioned<LedgerMetadata> writtenMetadata,
+                                  Throwable exception, LedgerMetadataBuilder metadataBuilder) {
         if (exception != null) {
             if (this.generateLedgerId
                 && (BKException.getExceptionCode(exception) == BKException.Code.LedgerExistException)) {
                 // retry to generate a new ledger id
-                generateLedgerIdAndCreateLedger();
+                generateLedgerIdAndCreateLedger(metadataBuilder);
             } else {
                 createComplete(BKException.getExceptionCode(exception), null);
             }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java

@@ -45,6 +45,7 @@
 @Unstable
 @VisibleForTesting
 public class LedgerMetadataBuilder {
+    private long ledgerId = -1L;
     private int metadataFormatVersion = CURRENT_METADATA_FORMAT_VERSION;
     private int ensembleSize = 3;
     private int writeQuorumSize = 3;
@@ -72,6 +73,7 @@ public static LedgerMetadataBuilder create() {
 
     public static LedgerMetadataBuilder from(LedgerMetadata other) {
         LedgerMetadataBuilder builder = new LedgerMetadataBuilder();
+        builder.ledgerId = other.getLedgerId();
         builder.metadataFormatVersion = other.getMetadataFormatVersion();
         builder.ensembleSize = other.getEnsembleSize();
         builder.writeQuorumSize = other.getWriteQuorumSize();
@@ -100,6 +102,11 @@ public static LedgerMetadataBuilder from(LedgerMetadata other) {
         return builder;
     }
 
+    public LedgerMetadataBuilder withId(long ledgerId) {
+        this.ledgerId = ledgerId;
+        return this;
+    }
+
     public LedgerMetadataBuilder withMetadataFormatVersion(int version) {
         this.metadataFormatVersion = version;
         return this;
@@ -190,10 +197,11 @@ public LedgerMetadataBuilder withCToken(long cToken) {
     }
 
     public LedgerMetadata build() {
+        checkArgument(ledgerId >= 0, "Ledger id must be set");
         checkArgument(ensembleSize >= writeQuorumSize, "Write quorum must be less or equal to ensemble size");
         checkArgument(writeQuorumSize >= ackQuorumSize, "Write quorum must be greater or equal to ack quorum");
 
-        return new LedgerMetadataImpl(metadataFormatVersion,
+        return new LedgerMetadataImpl(ledgerId, metadataFormatVersion,
                                       ensembleSize, writeQuorumSize, ackQuorumSize,
                                       state, lastEntryId, length, ensembles,
                                       digestType, password, ctime, storeCtime,

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataImpl.java

@@ -45,10 +45,14 @@
  *
  * <p>It provides parsing and serialization methods of such metadata.
  */
-@EqualsAndHashCode
+@EqualsAndHashCode(exclude =
+        "ledgerId" // ledgerId is not serialized inside ZK node data
+)
 class LedgerMetadataImpl implements LedgerMetadata {
     static final Logger LOG = LoggerFactory.getLogger(LedgerMetadataImpl.class);
 
+    private final long ledgerId;
+
     private final int metadataFormatVersion;
     private final int ensembleSize;
     private final int writeQuorumSize;
@@ -71,7 +75,8 @@ class LedgerMetadataImpl implements LedgerMetadata {
 
     private long cToken;
 
-    LedgerMetadataImpl(int metadataFormatVersion,
+    LedgerMetadataImpl(long ledgerId,
+                       int metadataFormatVersion,
                        int ensembleSize,
                        int writeQuorumSize,
                        int ackQuorumSize,
@@ -97,6 +102,7 @@ class LedgerMetadataImpl implements LedgerMetadata {
                       || (!digestType.isPresent() && !password.isPresent()),
                       "Either both password and digest type must be set, or neither");
 
+        this.ledgerId = ledgerId;
         this.metadataFormatVersion = metadataFormatVersion;
         this.ensembleSize = ensembleSize;
         this.writeQuorumSize = writeQuorumSize;
@@ -135,6 +141,11 @@ class LedgerMetadataImpl implements LedgerMetadata {
         this.customMetadata = ImmutableMap.copyOf(customMetadata);
     }
 
+    @Override
+    public long getLedgerId() {
+        return ledgerId;
+    }
+
     @Override
     public NavigableMap<Long, ? extends List<BookieId>> getAllEnsembles() {
         return ensembles;

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/BookKeeper.java

@@ -20,6 +20,7 @@
  */
 package org.apache.bookkeeper.client.api;
 
+import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.client.impl.BookKeeperBuilderImpl;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience.Public;
 import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
@@ -65,6 +66,21 @@ static BookKeeperBuilder newBuilder(final ClientConfiguration clientConfiguratio
      */
     DeleteBuilder newDeleteLedgerOp();
 
+    /**
+     * List ledgers.
+     *
+     * @return a builder useful to list ledgers.
+     */
+    ListLedgersResultBuilder newListLedgersOp();
+
+    /**
+     * Get ledger metadata of a given ledger id.
+     *
+     * @param ledgerId id of the ledger.
+     * @return a <code>CompletableFuture</code> instance containing ledger metadata.
+     */
+    CompletableFuture<LedgerMetadata> getLedgerMetadata(long ledgerId);
+
     /**
      * Close the client and release every resource.
      *

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/LedgerMetadata.java

@@ -34,6 +34,12 @@
 @LimitedPrivate
 @Unstable
 public interface LedgerMetadata {
+    /**
+     * Returns the id of this ledger.
+     *
+     * @return the id of this ledger.
+     */
+    long getLedgerId();
 
     /**
      * Returns the ensemble size of this ledger.

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/LedgersIterator.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2020 The Apache Software Foundation.
+ *
+ * Licensed 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.client.api;
+
+import java.io.IOException;
+
+/**
+ * Iterator for ledgers.
+ */
+public interface LedgersIterator {
+
+    /**
+     * Return true if there is at least one ledger to visit.
+     *
+     * @return true if there is at least one ledger to visit.
+     * @throws IOException thrown when there is a problem accessing the ledger metadata store.
+     */
+    boolean hasNext() throws IOException;
+
+    /**
+     * Return next ledger id.
+     *
+     * @return next ledger id.
+     * @throws IOException thrown when there is a problem accessing the ledger metadata store.
+     */
+    long next() throws IOException;
+
+}