You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/04/27 17:28:04 UTC
[GitHub] sijie closed pull request #1640: Read OffloadedLedger if available
sijie closed pull request #1640: Read OffloadedLedger if available
URL: https://github.com/apache/incubator-pulsar/pull/1640
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 197f0b747b..697f5eb716 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1358,23 +1358,32 @@ void asyncReadEntries(OpReadEntry opReadEntry) {
mbean.startDataLedgerOpenOp();
CompletableFuture<ReadHandle> promise = new CompletableFuture<>();
- bookKeeper.newOpenLedgerOp()
- .withRecovery(true)
- .withLedgerId(ledgerId)
- .withDigestType(config.getDigestType())
- .withPassword(config.getPassword()).execute()
- .whenCompleteAsync((res,ex) -> {
- mbean.endDataLedgerOpenOp();
- if (ex != null) {
- ledgerCache.remove(ledgerId, promise);
- promise.completeExceptionally(createManagedLedgerException(ex));
- } else {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Successfully opened ledger {} for reading", name, ledgerId);
- }
- promise.complete(res);
+
+ LedgerInfo info = ledgers.get(ledgerId);
+ CompletableFuture<ReadHandle> openFuture = new CompletableFuture<>();
+ if (info != null && info.hasOffloadContext() && info.getOffloadContext().getComplete()) {
+ UUID uid = new UUID(info.getOffloadContext().getUidMsb(),
+ info.getOffloadContext().getUidLsb());
+ openFuture = config.getLedgerOffloader().readOffloaded(ledgerId, uid);
+ } else {
+ openFuture = bookKeeper.newOpenLedgerOp()
+ .withRecovery(true)
+ .withLedgerId(ledgerId)
+ .withDigestType(config.getDigestType())
+ .withPassword(config.getPassword()).execute();
+ }
+ openFuture.whenCompleteAsync((res,ex) -> {
+ mbean.endDataLedgerOpenOp();
+ if (ex != null) {
+ ledgerCache.remove(ledgerId, promise);
+ promise.completeExceptionally(createManagedLedgerException(ex));
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Successfully opened ledger {} for reading", name, ledgerId);
}
- }, executor.chooseThread(name));
+ promise.complete(res);
+ }
+ }, executor.chooseThread(name));
return promise;
});
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java
new file mode 100644
index 0000000000..8bbb44ae4b
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java
@@ -0,0 +1,289 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.mledger.impl;
+
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+
+import io.netty.buffer.ByteBuf;
+
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.bookkeeper.client.api.DigestType;
+import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.impl.LedgerEntriesImpl;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class OffloadPrefixReadTest extends MockedBookKeeperTestCase {
+ private static final Logger log = LoggerFactory.getLogger(OffloadPrefixReadTest.class);
+
+ @Test
+ public void testOffloadRead() throws Exception {
+ MockLedgerOffloader offloader = spy(new MockLedgerOffloader());
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(10);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ config.setRetentionTime(10, TimeUnit.MINUTES);
+ config.setLedgerOffloader(offloader);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger", config);
+
+ for (int i = 0; i < 25; i++) {
+ String content = "entry-" + i;
+ ledger.addEntry(content.getBytes());
+ }
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 3);
+
+ ledger.offloadPrefix(ledger.getLastConfirmedEntry());
+
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 3);
+ Assert.assertEquals(ledger.getLedgersInfoAsList().stream()
+ .filter(e -> e.getOffloadContext().getComplete()).count(), 2);
+ Assert.assertTrue(ledger.getLedgersInfoAsList().get(0).getOffloadContext().getComplete());
+ Assert.assertTrue(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getComplete());
+
+ UUID firstLedgerUUID = new UUID(ledger.getLedgersInfoAsList().get(0).getOffloadContext().getUidMsb(),
+ ledger.getLedgersInfoAsList().get(0).getOffloadContext().getUidLsb());
+ UUID secondLedgerUUID = new UUID(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getUidMsb(),
+ ledger.getLedgersInfoAsList().get(1).getOffloadContext().getUidLsb());
+
+ ManagedCursor cursor = ledger.newNonDurableCursor(PositionImpl.earliest);
+ int i = 0;
+ for (Entry e : cursor.readEntries(10)) {
+ Assert.assertEquals(new String(e.getData()), "entry-" + i++);
+ }
+ verify(offloader, times(1)).readOffloaded(anyLong(), anyObject());
+ verify(offloader).readOffloaded(anyLong(), eq(firstLedgerUUID));
+
+ for (Entry e : cursor.readEntries(10)) {
+ Assert.assertEquals(new String(e.getData()), "entry-" + i++);
+ }
+ verify(offloader, times(2)).readOffloaded(anyLong(), anyObject());
+ verify(offloader).readOffloaded(anyLong(), eq(secondLedgerUUID));
+
+ for (Entry e : cursor.readEntries(5)) {
+ Assert.assertEquals(new String(e.getData()), "entry-" + i++);
+ }
+ verify(offloader, times(2)).readOffloaded(anyLong(), anyObject());
+ }
+
+ static class MockLedgerOffloader implements LedgerOffloader {
+ ConcurrentHashMap<UUID, ReadHandle> offloads = new ConcurrentHashMap<UUID, ReadHandle>();
+
+ @Override
+ public CompletableFuture<Void> offload(ReadHandle ledger,
+ UUID uuid,
+ Map<String, String> extraMetadata) {
+ CompletableFuture<Void> promise = new CompletableFuture<>();
+ try {
+ offloads.put(uuid, new MockOffloadReadHandle(ledger));
+ promise.complete(null);
+ } catch (Exception e) {
+ promise.completeExceptionally(e);
+ }
+ return promise;
+ }
+
+ @Override
+ public CompletableFuture<ReadHandle> readOffloaded(long ledgerId, UUID uuid) {
+ return CompletableFuture.completedFuture(offloads.get(uuid));
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteOffloaded(long ledgerId, UUID uuid) {
+ offloads.remove(uuid);
+ return CompletableFuture.completedFuture(null);
+ };
+ }
+
+ static class MockOffloadReadHandle implements ReadHandle {
+ final long id;
+ final List<ByteBuf> entries = Lists.newArrayList();
+ final LedgerMetadata metadata;
+
+ MockOffloadReadHandle(ReadHandle toCopy) throws Exception {
+ id = toCopy.getId();
+ long lac = toCopy.getLastAddConfirmed();
+ try (LedgerEntries entries = toCopy.read(0, lac)) {
+ for (LedgerEntry e : entries) {
+ this.entries.add(e.getEntryBuffer().retainedSlice());
+ }
+ }
+ metadata = new MockMetadata(toCopy.getLedgerMetadata());
+ }
+
+ @Override
+ public long getId() { return id; }
+
+ @Override
+ public LedgerMetadata getLedgerMetadata() {
+ return metadata;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntry) {
+ List<LedgerEntry> readEntries = Lists.newArrayList();
+ for (long eid = firstEntry; eid <= lastEntry; eid++) {
+ ByteBuf buf = entries.get((int)eid).retainedSlice();
+ readEntries.add(LedgerEntryImpl.create(id, eid, buf.readableBytes(), buf));
+ }
+ return CompletableFuture.completedFuture(LedgerEntriesImpl.create(readEntries));
+ }
+
+ @Override
+ public CompletableFuture<LedgerEntries> readUnconfirmedAsync(long firstEntry, long lastEntry) {
+ return unsupported();
+ }
+
+ @Override
+ public CompletableFuture<Long> readLastAddConfirmedAsync() {
+ return unsupported();
+ }
+
+ @Override
+ public CompletableFuture<Long> tryReadLastAddConfirmedAsync() {
+ return unsupported();
+ }
+
+ @Override
+ public long getLastAddConfirmed() {
+ return entries.size() - 1;
+ }
+
+ @Override
+ public long getLength() {
+ return metadata.getLength();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return metadata.isClosed();
+ }
+
+ @Override
+ public CompletableFuture<LastConfirmedAndEntry> readLastAddConfirmedAndEntryAsync(long entryId,
+ long timeOutInMillis,
+ boolean parallel) {
+ return unsupported();
+ }
+
+ private <T> CompletableFuture<T> unsupported() {
+ CompletableFuture<T> future = new CompletableFuture<>();
+ future.completeExceptionally(new UnsupportedOperationException());
+ return future;
+ }
+ }
+
+ static class MockMetadata implements LedgerMetadata {
+ private final int ensembleSize;
+ private final int writeQuorumSize;
+ private final int ackQuorumSize;
+ private final long lastEntryId;
+ private final long length;
+ private final DigestType digestType;
+ private final long ctime;
+ private final boolean isClosed;
+ private final Map<String, byte[]> customMetadata;
+
+ MockMetadata(LedgerMetadata toCopy) {
+ ensembleSize = toCopy.getEnsembleSize();
+ writeQuorumSize = toCopy.getWriteQuorumSize();
+ ackQuorumSize = toCopy.getAckQuorumSize();
+ lastEntryId = toCopy.getLastEntryId();
+ length = toCopy.getLength();
+ digestType = toCopy.getDigestType();
+ ctime = toCopy.getCtime();
+ isClosed = toCopy.isClosed();
+
+ customMetadata = ImmutableMap.copyOf(toCopy.getCustomMetadata());
+ }
+
+ @Override
+ public int getEnsembleSize() { return ensembleSize; }
+
+ @Override
+ public int getWriteQuorumSize() { return writeQuorumSize; }
+
+ @Override
+ public int getAckQuorumSize() { return ackQuorumSize; }
+
+ @Override
+ public long getLastEntryId() { return lastEntryId; }
+
+ @Override
+ public long getLength() { return length; }
+
+ @Override
+ public DigestType getDigestType() { return digestType; }
+
+ @Override
+ public long getCtime() { return ctime; }
+
+ @Override
+ public boolean isClosed() { return isClosed; }
+
+ @Override
+ public Map<String, byte[]> getCustomMetadata() { return customMetadata; }
+
+ @Override
+ public List<BookieSocketAddress> getEnsembleAt(long entryId) {
+ throw new UnsupportedOperationException("Pulsar shouldn't look at this");
+ }
+
+ @Override
+ public NavigableMap<Long, ? extends List<BookieSocketAddress>> getAllEnsembles() {
+ throw new UnsupportedOperationException("Pulsar shouldn't look at this");
+ }
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services