This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ab67386 Managed ledger uses ReadHandle in read path (#1513)
ab67386 is described below
commit ab67386b966ab92b55e70bef4a48d265a07f509a
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Tue Apr 10 18:56:45 2018 +0200
Managed ledger uses ReadHandle in read path (#1513)
* Managed ledger uses ReadHandle in read path
BookKeeper 4.6 introduced a new API for reading, called
ReadHandle. This API is an small interface, unlike LedgerHandle, so it
is possible to provide implementations of ReadHandle that read from
sources other than bookkeeper, while presenting the same interface as
when reading from bookkeeper.
This patch changes the managed ledger entry read path to use
ReadHandle. This is rquired to implement tiered storage (PIP-17).
Master issue: 1511
* Use whenCompleteAsync and chooseThread where possible
To avoid creating yet another runnable.
* Fix failing BrokerBkEnsembleTest
And another merge conflict.
---
.../bookkeeper/mledger/ManagedLedgerConfig.java | 2 +-
.../apache/bookkeeper/mledger/impl/EntryCache.java | 6 +-
.../bookkeeper/mledger/impl/EntryCacheImpl.java | 136 +++++++++++----------
.../bookkeeper/mledger/impl/EntryCacheManager.java | 56 ++++-----
.../apache/bookkeeper/mledger/impl/EntryImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 6 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 75 +++++++-----
.../mledger/impl/ManagedLedgerOfflineBacklog.java | 5 +-
.../bookkeeper/mledger/impl/EntryCacheTest.java | 56 ++++-----
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 3 +-
.../mledger/impl/ManagedLedgerBkTest.java | 9 +-
.../mledger/impl/ManagedLedgerErrorsTest.java | 2 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 2 +-
.../PulsarConfigurationLoaderTest.java | 2 +-
.../service/schema/BookkeeperSchemaStorage.java | 4 +-
.../pulsar/testclient/ManagedLedgerWriter.java | 2 +-
16 files changed, 190 insertions(+), 178 deletions(-)
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index b38999b..9aa5e1d 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -24,7 +24,7 @@ import com.google.common.annotations.Beta;
import com.google.common.base.Charsets;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.DigestType;
/**
* Configuration class for a ManagedLedger.
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java
index bb8f2a6..0c99650 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java
@@ -18,7 +18,7 @@
*/
package org.apache.bookkeeper.mledger.impl;
-import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
import org.apache.commons.lang3.tuple.Pair;
@@ -94,7 +94,7 @@ public interface EntryCache extends Comparable<EntryCache> {
* @param ctx
* the context object
*/
- void asyncReadEntry(LedgerHandle lh, long firstEntry, long lastEntry,
boolean isSlowestReader,
+ void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry,
boolean isSlowestReader,
ReadEntriesCallback callback, Object ctx);
/**
@@ -111,7 +111,7 @@ public interface EntryCache extends Comparable<EntryCache> {
* @param ctx
* the context object
*/
- void asyncReadEntry(LedgerHandle lh, PositionImpl position,
ReadEntryCallback callback, Object ctx);
+ void asyncReadEntry(ReadHandle lh, PositionImpl position,
ReadEntryCallback callback, Object ctx);
/**
* Get the total size in bytes of all the entries stored in this cache.
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
index 37ddc54..ff80fec 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
@@ -28,10 +28,11 @@ import com.google.common.primitives.Longs;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.LedgerEntry;
-import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.BKException;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -159,7 +160,7 @@ public class EntryCacheImpl implements EntryCache {
}
@Override
- public void asyncReadEntry(LedgerHandle lh, PositionImpl position, final
ReadEntryCallback callback,
+ public void asyncReadEntry(ReadHandle lh, PositionImpl position, final
ReadEntryCallback callback,
final Object ctx) {
if (log.isDebugEnabled()) {
log.debug("[{}] Reading entry ledger {}: {}", ml.getName(),
lh.getId(), position.getEntryId());
@@ -171,37 +172,38 @@ public class EntryCacheImpl implements EntryCache {
manager.mlFactoryMBean.recordCacheHit(cachedEntry.getLength());
callback.readEntryComplete(cachedEntry, ctx);
} else {
- lh.asyncReadEntries(position.getEntryId(), position.getEntryId(),
(rc, ledgerHandle, sequence, obj) -> {
- if (rc != BKException.Code.OK) {
- ml.invalidateLedgerHandle(ledgerHandle, rc);
- callback.readEntryFailed(createManagedLedgerException(rc),
obj);
- return;
- }
-
- if (sequence.hasMoreElements()) {
- LedgerEntry ledgerEntry = sequence.nextElement();
- EntryImpl returnEntry = EntryImpl.create(ledgerEntry);
-
- // The EntryImpl is now the owner of the buffer, so we can
release the original one
- ledgerEntry.getEntryBuffer().release();
-
- manager.mlFactoryMBean.recordCacheMiss(1,
returnEntry.getLength());
- ml.mbean.addReadEntriesSample(1, returnEntry.getLength());
-
- ml.getExecutor().executeOrdered(ml.getName(), safeRun(()
-> {
- callback.readEntryComplete(returnEntry, obj);
- }));
- } else {
- // got an empty sequence
- callback.readEntryFailed(new ManagedLedgerException("Could
not read given position"), obj);
- }
- }, ctx);
+ lh.readAsync(position.getEntryId(),
position.getEntryId()).whenCompleteAsync(
+ (ledgerEntries, exception) -> {
+ if (exception != null) {
+ ml.invalidateLedgerHandle(lh, exception);
+
callback.readEntryFailed(createManagedLedgerException(exception), ctx);
+ return;
+ }
+
+ try {
+ Iterator<LedgerEntry> iterator =
ledgerEntries.iterator();
+ if (iterator.hasNext()) {
+ LedgerEntry ledgerEntry = iterator.next();
+ EntryImpl returnEntry =
EntryImpl.create(ledgerEntry);
+
+ manager.mlFactoryMBean.recordCacheMiss(1,
returnEntry.getLength());
+ ml.mbean.addReadEntriesSample(1,
returnEntry.getLength());
+ callback.readEntryComplete(returnEntry, ctx);
+ } else {
+ // got an empty sequence
+ callback.readEntryFailed(new
ManagedLedgerException("Could not read given position"),
+ ctx);
+ }
+ } finally {
+ ledgerEntries.close();
+ }
+ }, ml.getExecutor().chooseThread(ml.getName()));
}
}
@Override
@SuppressWarnings({ "unchecked", "rawtypes" })
- public void asyncReadEntry(LedgerHandle lh, long firstEntry, long
lastEntry, boolean isSlowestReader,
+ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry,
boolean isSlowestReader,
final ReadEntriesCallback callback, Object ctx) {
final long ledgerId = lh.getId();
final int entriesToRead = (int) (lastEntry - firstEntry) + 1;
@@ -239,43 +241,43 @@ public class EntryCacheImpl implements EntryCache {
}
// Read all the entries from bookkeeper
- lh.asyncReadEntries(firstEntry, lastEntry, (rc, lh1, sequence, cb)
-> {
-
- if (rc != BKException.Code.OK) {
- if (rc == BKException.Code.TooManyRequestsException) {
-
callback.readEntriesFailed(createManagedLedgerException(rc), ctx);
- } else {
- ml.invalidateLedgerHandle(lh1, rc);
- ManagedLedgerException mlException =
createManagedLedgerException(rc);
- callback.readEntriesFailed(mlException, ctx);
- }
- return;
- }
-
- checkNotNull(ml.getName());
- checkNotNull(ml.getExecutor());
- ml.getExecutor().executeOrdered(ml.getName(), safeRun(() -> {
- // We got the entries, we need to transform them to a
List<> type
- long totalSize = 0;
- final List<EntryImpl> entriesToReturn =
Lists.newArrayListWithExpectedSize(entriesToRead);
- while (sequence.hasMoreElements()) {
- // Insert the entries at the end of the list (they
will be unsorted for now)
- LedgerEntry ledgerEntry = sequence.nextElement();
- EntryImpl entry = EntryImpl.create(ledgerEntry);
- ledgerEntry.getEntryBuffer().release();
-
- entriesToReturn.add(entry);
-
- totalSize += entry.getLength();
-
- }
-
-
manager.mlFactoryMBean.recordCacheMiss(entriesToReturn.size(), totalSize);
- ml.getMBean().addReadEntriesSample(entriesToReturn.size(),
totalSize);
-
- callback.readEntriesComplete((List) entriesToReturn, ctx);
- }));
- }, callback);
+ lh.readAsync(firstEntry, lastEntry).whenCompleteAsync(
+ (ledgerEntries, exception) -> {
+ if (exception != null) {
+ if (exception instanceof BKException
+ && ((BKException)exception).getCode() ==
BKException.Code.TooManyRequestsException) {
+
callback.readEntriesFailed(createManagedLedgerException(exception), ctx);
+ } else {
+ ml.invalidateLedgerHandle(lh, exception);
+ ManagedLedgerException mlException =
createManagedLedgerException(exception);
+ callback.readEntriesFailed(mlException, ctx);
+ }
+ return;
+ }
+
+ checkNotNull(ml.getName());
+ checkNotNull(ml.getExecutor());
+
+ try {
+ // We got the entries, we need to transform them
to a List<> type
+ long totalSize = 0;
+ final List<EntryImpl> entriesToReturn
+ =
Lists.newArrayListWithExpectedSize(entriesToRead);
+ for (LedgerEntry e : ledgerEntries) {
+ EntryImpl entry = EntryImpl.create(e);
+
+ entriesToReturn.add(entry);
+ totalSize += entry.getLength();
+ }
+
+
manager.mlFactoryMBean.recordCacheMiss(entriesToReturn.size(), totalSize);
+
ml.getMBean().addReadEntriesSample(entriesToReturn.size(), totalSize);
+
+ callback.readEntriesComplete((List)
entriesToReturn, ctx);
+ } finally {
+ ledgerEntries.close();
+ }
+ }, ml.getExecutor().chooseThread(ml.getName()));
}
}
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
index 9f6f837..c551002 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
@@ -33,8 +33,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.LedgerEntry;
-import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
@@ -190,37 +190,35 @@ public class EntryCacheManager {
}
@Override
- public void asyncReadEntry(LedgerHandle lh, long firstEntry, long
lastEntry, boolean isSlowestReader,
+ public void asyncReadEntry(ReadHandle lh, long firstEntry, long
lastEntry, boolean isSlowestReader,
final ReadEntriesCallback callback, Object ctx) {
- lh.asyncReadEntries(firstEntry, lastEntry, new ReadCallback() {
- public void readComplete(int rc, LedgerHandle lh,
Enumeration<LedgerEntry> seq, Object bkctx) {
- if (rc != BKException.Code.OK) {
-
callback.readEntriesFailed(createManagedLedgerException(rc), ctx);
- return;
- }
-
- List<Entry> entries = Lists.newArrayList();
- long totalSize = 0;
- while (seq.hasMoreElements()) {
- // Insert the entries at the end of the list (they
will be unsorted for now)
- LedgerEntry ledgerEntry = seq.nextElement();
- EntryImpl entry = EntryImpl.create(ledgerEntry);
- ledgerEntry.getEntryBuffer().release();
-
- entries.add(entry);
- totalSize += entry.getLength();
- }
-
- mlFactoryMBean.recordCacheMiss(entries.size(), totalSize);
- ml.mbean.addReadEntriesSample(entries.size(), totalSize);
-
- callback.readEntriesComplete(entries, null);
- }
- }, null);
+ lh.readAsync(firstEntry, lastEntry).whenComplete(
+ (ledgerEntries, exception) -> {
+ if (exception != null) {
+
callback.readEntriesFailed(createManagedLedgerException(exception), ctx);
+ return;
+ }
+ List<Entry> entries = Lists.newArrayList();
+ long totalSize = 0;
+ try {
+ for (LedgerEntry e : ledgerEntries) {
+ // Insert the entries at the end of the list
(they will be unsorted for now)
+ EntryImpl entry = EntryImpl.create(e);
+ entries.add(entry);
+ totalSize += entry.getLength();
+ }
+ } finally {
+ ledgerEntries.close();
+ }
+ mlFactoryMBean.recordCacheMiss(entries.size(),
totalSize);
+ ml.mbean.addReadEntriesSample(entries.size(),
totalSize);
+
+ callback.readEntriesComplete(entries, null);
+ });
}
@Override
- public void asyncReadEntry(LedgerHandle lh, PositionImpl position,
AsyncCallbacks.ReadEntryCallback callback,
+ public void asyncReadEntry(ReadHandle lh, PositionImpl position,
AsyncCallbacks.ReadEntryCallback callback,
Object ctx) {
}
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
index d1c6def..bf3e925 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
@@ -25,7 +25,7 @@ import io.netty.util.AbstractReferenceCounted;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCounted;
-import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.mledger.Entry;
public final class EntryImpl extends AbstractReferenceCounted implements
Entry, Comparable<EntryImpl>, ReferenceCounted {
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 443a873..b043dc8 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -92,6 +92,7 @@ public class ManagedCursorImpl implements ManagedCursor {
protected final ManagedLedgerConfig config;
protected final ManagedLedgerImpl ledger;
private final String name;
+ private final BookKeeper.DigestType digestType;
protected volatile PositionImpl markDeletePosition;
protected volatile PositionImpl readPosition;
@@ -180,6 +181,7 @@ public class ManagedCursorImpl implements ManagedCursor {
this.config = config;
this.ledger = ledger;
this.name = cursorName;
+ this.digestType =
BookKeeper.DigestType.fromApiDigestType(config.getDigestType());
STATE_UPDATER.set(this, State.Uninitialized);
PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.set(this, 0);
PENDING_READ_OPS_UPDATER.set(this, 0);
@@ -254,7 +256,7 @@ public class ManagedCursorImpl implements ManagedCursor {
// a new ledger and write the position into it
ledger.mbean.startCursorLedgerOpenOp();
long ledgerId = info.getCursorsLedgerId();
- bookkeeper.asyncOpenLedger(ledgerId, config.getDigestType(),
config.getPassword(), (rc, lh, ctx) -> {
+ bookkeeper.asyncOpenLedger(ledgerId, digestType, config.getPassword(),
(rc, lh, ctx) -> {
if (log.isDebugEnabled()) {
log.debug("[{}] Opened ledger {} for consumer {}. rc={}",
ledger.getName(), ledgerId, name, rc);
}
@@ -1922,7 +1924,7 @@ public class ManagedCursorImpl implements ManagedCursor {
ledger.mbean.startCursorLedgerCreateOp();
bookkeeper.asyncCreateLedger(config.getMetadataEnsemblesize(),
config.getMetadataWriteQuorumSize(),
- config.getMetadataAckQuorumSize(), config.getDigestType(),
config.getPassword(), (rc, lh, ctx) -> {
+ config.getMetadataAckQuorumSize(), digestType,
config.getPassword(), (rc, lh, ctx) -> {
ledger.getExecutor().execute(safeRun(() -> {
ledger.mbean.endCursorLedgerCreateOp();
if (rc != BKException.Code.OK) {
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 f6c7e3f..77b1dff 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
@@ -54,6 +54,8 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.client.api.DigestType;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
@@ -104,11 +106,12 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
private final BookKeeper bookKeeper;
private final String name;
+ private final BookKeeper.DigestType digestType;
private ManagedLedgerConfig config;
private final MetaStore store;
- private final ConcurrentLongHashMap<CompletableFuture<LedgerHandle>>
ledgerCache = new ConcurrentLongHashMap<>();
+ private final ConcurrentLongHashMap<CompletableFuture<ReadHandle>>
ledgerCache = new ConcurrentLongHashMap<>();
private final NavigableMap<Long, LedgerInfo> ledgers = new
ConcurrentSkipListMap<>();
private volatile Stat ledgersStat;
@@ -212,6 +215,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
this.config = config;
this.store = store;
this.name = name;
+ this.digestType =
BookKeeper.DigestType.fromApiDigestType(config.getDigestType());
this.scheduledExecutor = scheduledExecutor;
this.executor = orderedExecutor;
TOTAL_SIZE_UPDATER.set(this, 0);
@@ -278,7 +282,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
log.debug("[{}] Opening legder {}", name, id);
}
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, config.getDigestType(),
config.getPassword(), opencb, null);
+ bookKeeper.asyncOpenLedger(id, digestType,
config.getPassword(), opencb, null);
} else {
initializeBookKeeper(callback);
}
@@ -342,7 +346,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
this.lastLedgerCreationInitiationTimestamp = System.nanoTime();
mbean.startDataLedgerCreateOp();
bookKeeper.asyncCreateLedger(config.getEnsembleSize(),
config.getWriteQuorumSize(), config.getAckQuorumSize(),
- config.getDigestType(), config.getPassword(), (rc, lh, ctx) ->
{
+ digestType, config.getPassword(), (rc, lh, ctx) -> {
executor.executeOrdered(name, safeRun(() -> {
mbean.endDataLedgerCreateOp();
if (rc != BKException.Code.OK) {
@@ -528,7 +532,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
this.lastLedgerCreationInitiationTimestamp = System.nanoTime();
mbean.startDataLedgerCreateOp();
bookKeeper.asyncCreateLedger(config.getEnsembleSize(),
config.getWriteQuorumSize(),
- config.getAckQuorumSize(), config.getDigestType(),
config.getPassword(), this, null,
+ config.getAckQuorumSize(), digestType,
config.getPassword(), this, null,
Collections.emptyMap());
}
} else {
@@ -1263,7 +1267,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
this.lastLedgerCreationInitiationTimestamp = System.nanoTime();
mbean.startDataLedgerCreateOp();
bookKeeper.asyncCreateLedger(config.getEnsembleSize(),
config.getWriteQuorumSize(),
- config.getAckQuorumSize(), config.getDigestType(),
config.getPassword(), this, null,
+ config.getAckQuorumSize(), digestType,
config.getPassword(), this, null,
Collections.emptyMap());
}
}
@@ -1314,52 +1318,53 @@ public class ManagedLedgerImpl implements
ManagedLedger, CreateCallback {
}
}
- CompletableFuture<LedgerHandle> getLedgerHandle(long ledgerId) {
- CompletableFuture<LedgerHandle> ledgerHandle =
ledgerCache.get(ledgerId);
+ CompletableFuture<ReadHandle> getLedgerHandle(long ledgerId) {
+ CompletableFuture<ReadHandle> ledgerHandle = ledgerCache.get(ledgerId);
if (ledgerHandle != null) {
return ledgerHandle;
}
// If not present try again and create if necessary
return ledgerCache.computeIfAbsent(ledgerId, lid -> {
- // Open the ledger for reading if it was not already opened
- CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] Asynchronously opening ledger {} for read",
name, ledgerId);
- }
- mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(ledgerId, config.getDigestType(),
config.getPassword(),
- (int rc, LedgerHandle lh, Object ctx) -> {
- executor.executeOrdered(name, safeRun(() -> {
+ // Open the ledger for reading if it was not already opened
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Asynchronously opening ledger {} for
read", name, ledgerId);
+ }
+ 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 (rc != BKException.Code.OK) {
- // Remove the ledger future from cache to give
chance to reopen it later
- ledgerCache.remove(ledgerId, future);
-
future.completeExceptionally(createManagedLedgerException(rc));
+ if (ex != null) {
+ ledgerCache.remove(ledgerId, promise);
+
promise.completeExceptionally(createManagedLedgerException(ex));
} else {
if (log.isDebugEnabled()) {
- log.debug("[{}] Successfully opened ledger
{} for reading", name, lh.getId());
+ log.debug("[{}] Successfully opened ledger
{} for reading", name, ledgerId);
}
- future.complete(lh);
+ promise.complete(res);
}
- }));
- }, null);
- return future;
- });
+ }, executor.chooseThread(name));
+ return promise;
+ });
}
- void invalidateLedgerHandle(LedgerHandle ledgerHandle, int rc) {
+ void invalidateLedgerHandle(ReadHandle ledgerHandle, Throwable t) {
long ledgerId = ledgerHandle.getId();
if (ledgerId != currentLedger.getId()) {
// remove handle from ledger cache since we got a (read) error
ledgerCache.remove(ledgerId);
if (log.isDebugEnabled()) {
- log.debug("[{}] Removed ledger {} from cache (after read
error: {})", name, ledgerId, rc);
+ log.debug("[{}] Removed ledger {} from cache (after read
error)", name, ledgerId, t);
}
} else {
if (log.isDebugEnabled()) {
- log.debug("[{}] Ledger that encountered read error {} is
current ledger", name, rc);
+ log.debug("[{}] Ledger that encountered read error is current
ledger", name, t);
}
}
}
@@ -1384,7 +1389,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
}
- private void internalReadFromLedger(LedgerHandle ledger, OpReadEntry
opReadEntry) {
+ private void internalReadFromLedger(ReadHandle ledger, OpReadEntry
opReadEntry) {
// Perform the read
long firstEntry = opReadEntry.readPosition.getEntryId();
@@ -2256,6 +2261,14 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
}
}
+ public static ManagedLedgerException
createManagedLedgerException(Throwable t) {
+ if (t instanceof org.apache.bookkeeper.client.api.BKException) {
+ return
createManagedLedgerException(((org.apache.bookkeeper.client.api.BKException)t).getCode());
+ } else {
+ return new ManagedLedgerException("Unknown exception");
+ }
+ }
+
private static final Logger log =
LoggerFactory.getLogger(ManagedLedgerImpl.class);
}
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerOfflineBacklog.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerOfflineBacklog.java
index 39405af..06acb5b 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerOfflineBacklog.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerOfflineBacklog.java
@@ -34,6 +34,7 @@ import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeperAdmin;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.pulsar.common.naming.TopicName;
@@ -52,9 +53,9 @@ public class ManagedLedgerOfflineBacklog {
private boolean accurate = false;
private String brokerName;
- public ManagedLedgerOfflineBacklog(BookKeeper.DigestType digestType,
byte[] password, String brokerName,
+ public ManagedLedgerOfflineBacklog(DigestType digestType, byte[] password,
String brokerName,
boolean accurate) {
- this.digestType = digestType;
+ this.digestType = BookKeeper.DigestType.fromApiDigestType(digestType);
this.password = password;
this.accurate = accurate;
this.brokerName = brokerName;
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheTest.java
index 1be2692..c48ea24 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheTest.java
@@ -27,11 +27,13 @@ import io.netty.buffer.Unpooled;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Vector;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
-import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.LedgerEntry;
-import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.BKException.BKNoSuchLedgerExistsException;
+import org.apache.bookkeeper.client.api.BKException;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -59,7 +61,7 @@ public class EntryCacheTest extends MockedBookKeeperTestCase {
@Test(timeOut = 5000)
void testRead() throws Exception {
- LedgerHandle lh = getLedgerHandle();
+ ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
EntryCacheManager cacheManager = factory.getEntryCacheManager();
@@ -86,12 +88,12 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
counter.await();
// Verify no entries were read from bookkeeper
- verify(lh, never()).asyncReadEntries(anyLong(), anyLong(),
any(ReadCallback.class), any());
+ verify(lh, never()).readAsync(anyLong(), anyLong());
}
@Test(timeOut = 5000)
void testReadMissingBefore() throws Exception {
- LedgerHandle lh = getLedgerHandle();
+ ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
EntryCacheManager cacheManager = factory.getEntryCacheManager();
@@ -119,7 +121,7 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
@Test(timeOut = 5000)
void testReadMissingAfter() throws Exception {
- LedgerHandle lh = getLedgerHandle();
+ ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
EntryCacheManager cacheManager = factory.getEntryCacheManager();
@@ -147,7 +149,7 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
@Test(timeOut = 5000)
void testReadMissingMiddle() throws Exception {
- LedgerHandle lh = getLedgerHandle();
+ ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
EntryCacheManager cacheManager = factory.getEntryCacheManager();
@@ -176,7 +178,7 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
@Test(timeOut = 5000)
void testReadMissingMultiple() throws Exception {
- LedgerHandle lh = getLedgerHandle();
+ ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
EntryCacheManager cacheManager = factory.getEntryCacheManager();
@@ -205,18 +207,14 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
@Test(timeOut = 5000)
void testReadWithError() throws Exception {
- final LedgerHandle lh = getLedgerHandle();
+ final ReadHandle lh = getLedgerHandle();
when(lh.getId()).thenReturn((long) 0);
- doAnswer(new Answer<Object>() {
- public Object answer(InvocationOnMock invocation) {
- Object[] args = invocation.getArguments();
- ReadCallback callback = (ReadCallback) args[2];
- Object ctx = args[3];
-
callback.readComplete(BKException.Code.NoSuchLedgerExistsException, lh, null,
ctx);
- return null;
- }
- }).when(lh).asyncReadEntries(anyLong(), anyLong(),
any(ReadCallback.class), any());
+ doAnswer((invocation) -> {
+ CompletableFuture<LedgerEntries> future = new
CompletableFuture<>();
+ future.completeExceptionally(new
BKNoSuchLedgerExistsException());
+ return future;
+ }).when(lh).readAsync(anyLong(), anyLong());
EntryCacheManager cacheManager = factory.getEntryCacheManager();
EntryCache entryCache = cacheManager.getEntryCache(ml);
@@ -238,29 +236,25 @@ public class EntryCacheTest extends
MockedBookKeeperTestCase {
counter.await();
}
- private static LedgerHandle getLedgerHandle() {
- final LedgerHandle lh = mock(LedgerHandle.class);
+ private static ReadHandle getLedgerHandle() {
+ final ReadHandle lh = mock(ReadHandle.class);
final LedgerEntry ledgerEntry = mock(LedgerEntry.class,
Mockito.CALLS_REAL_METHODS);
- doReturn(new byte[10]).when(ledgerEntry).getEntry();
doReturn(Unpooled.wrappedBuffer(new
byte[10])).when(ledgerEntry).getEntryBuffer();
doReturn((long) 10).when(ledgerEntry).getLength();
- doAnswer(new Answer<Object>() {
- public Object answer(InvocationOnMock invocation) {
+ doAnswer((invocation) -> {
Object[] args = invocation.getArguments();
long firstEntry = (Long) args[0];
long lastEntry = (Long) args[1];
- ReadCallback callback = (ReadCallback) args[2];
- Object ctx = args[3];
Vector<LedgerEntry> entries = new Vector<LedgerEntry>();
for (int i = 0; i <= (lastEntry - firstEntry); i++) {
entries.add(ledgerEntry);
}
- callback.readComplete(0, lh, entries.elements(), ctx);
- return null;
- }
- }).when(lh).asyncReadEntries(anyLong(), anyLong(),
any(ReadCallback.class), any());
+ LedgerEntries ledgerEntries = mock(LedgerEntries.class);
+ doAnswer((invocation2) ->
entries.iterator()).when(ledgerEntries).iterator();
+ return CompletableFuture.completedFuture(ledgerEntries);
+ }).when(lh).readAsync(anyLong(), anyLong());
return lh;
}
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index ad85a35..8cf5ee4 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -2473,7 +2473,8 @@ public class ManagedCursorTest extends
MockedBookKeeperTestCase {
latch2.await();
try {
- bkc.openLedgerNoRecovery(ledgerId, mlConfig.getDigestType(),
mlConfig.getPassword());
+ bkc.openLedgerNoRecovery(ledgerId,
DigestType.fromApiDigestType(mlConfig.getDigestType()),
+ mlConfig.getPassword());
fail("ledger should have deleted due to update-cursor failure");
} catch (BKException e) {
// ok
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
index 3035c28..a5b58d7 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
@@ -33,8 +33,9 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeperTestClient;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
import org.apache.bookkeeper.mledger.Entry;
@@ -345,7 +346,7 @@ public class ManagedLedgerBkTest extends
BookKeeperClusterTestCase {
PositionImpl p1 = (PositionImpl) ledger.addEntry("entry-1".getBytes());
// Trigger the closure of the data ledger
- bkc.openLedger(p1.getLedgerId(), DigestType.CRC32C, new byte[] {});
+ bkc.openLedger(p1.getLedgerId(), BookKeeper.DigestType.CRC32C, new
byte[] {});
ledger.addEntry("entry-2".getBytes());
@@ -423,8 +424,8 @@ public class ManagedLedgerBkTest extends
BookKeeperClusterTestCase {
entries.forEach(e -> e.release());
ledger.close();
- ManagedLedgerOfflineBacklog offlineTopicBacklog = new
ManagedLedgerOfflineBacklog(DigestType.CRC32,
- "".getBytes(Charsets.UTF_8), "", false);
+ ManagedLedgerOfflineBacklog offlineTopicBacklog = new
ManagedLedgerOfflineBacklog(
+ DigestType.CRC32, "".getBytes(Charsets.UTF_8), "", false);
PersistentOfflineTopicStats offlineTopicStats =
offlineTopicBacklog.getEstimatedUnloadedTopicBacklog(
(ManagedLedgerFactoryImpl) factory,
"property/cluster/namespace/my-ledger");
factory.shutdown();
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
index 4efeefb..35eb986 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
@@ -24,7 +24,7 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
import org.apache.bookkeeper.mledger.Entry;
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index dd9cabd..1002714 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -24,7 +24,7 @@ import java.util.Optional;
import java.util.Properties;
import java.util.Set;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
import org.apache.pulsar.common.configuration.FieldContext;
import org.apache.pulsar.common.configuration.PulsarConfiguration;
diff --git
a/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
index 038fb4d..d830c4f 100644
---
a/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
+++
b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
@@ -32,7 +32,7 @@ import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Properties;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.testng.annotations.Test;
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index 3730121..c07217c 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -384,7 +384,7 @@ public class BookkeeperSchemaStorage implements
SchemaStorage {
config.getManagedLedgerDefaultEnsembleSize(),
config.getManagedLedgerDefaultWriteQuorum(),
config.getManagedLedgerDefaultAckQuorum(),
- config.getManagedLedgerDigestType(),
+
BookKeeper.DigestType.fromApiDigestType(config.getManagedLedgerDigestType()),
LedgerPassword,
(rc, handle, ctx) -> {
if (rc != BKException.Code.OK) {
@@ -402,7 +402,7 @@ public class BookkeeperSchemaStorage implements
SchemaStorage {
final CompletableFuture<LedgerHandle> future = new
CompletableFuture<>();
bookKeeper.asyncOpenLedger(
ledgerId,
- config.getManagedLedgerDigestType(),
+
BookKeeper.DigestType.fromApiDigestType(config.getManagedLedgerDigestType()),
LedgerPassword,
(rc, handle, ctx) -> {
if (rc != BKException.Code.OK) {
diff --git
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
index e975b37..0ffea79 100644
---
a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
+++
b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
@@ -49,7 +49,7 @@ import java.util.stream.Collectors;
import org.HdrHistogram.Histogram;
import org.HdrHistogram.Recorder;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
--
To stop receiving notification emails like this one, please contact
mme...@apache.org.
