This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new badd5f854a Pin a ledger's callbacks to a caller-chosen worker thread
via withOrderingKey (#4881)
badd5f854a is described below
commit badd5f854a80c0a03b90e0b06ff9fc7eb90382e8
Author: Matteo Merli <[email protected]>
AuthorDate: Thu Sep 10 07:55:36 2026 -0700
Pin a ledger's callbacks to a caller-chosen worker thread via
withOrderingKey (#4881)
* Client: option to pin a ledger's callbacks to a caller-chosen worker
thread
Add CreateBuilder/CreateAdvBuilder/OpenBuilder.withOrderingKey(Object). When
set, LedgerHandle.executor is mainWorkerPool.chooseThread(key) instead of
chooseThread(ledgerId), and every place that used to pick a thread by
ledger id
for the handle (handle callbacks, read op submission, speculative requests,
metadata updates, open and recovery completion) goes through that executor.
Bookie response dispatch follows the same thread: BookieClient gains an
Executor callbackExecutor overload for addEntry, readEntry,
batchReadEntries,
readLac, writeLac, forceLedger and readEntryWaitForLACUpdate, carried
through
BookieClientImpl, PerChannelBookieClient and the CompletionValue hierarchy
so
that responses (V2 and V3), connection failures, error-outs and timeouts all
run on the handle's thread. The previous signatures remain as default
methods.
Without a key nothing changes: null resolves to the ledger-id thread exactly
as before. This lets Pulsar pass its managed-ledger name as the key so the
ledger thread and the managed-ledger thread coincide, removing a
cross-thread
hop per add completion and per bookie response.
* Comment why LedgerHandle never boxes the ledger id when choosing its
thread
chooseThread(long) hashes the raw id while chooseThread(Object) goes through
hashCode(), and Long.hashCode folds the high 32 bits. Resolving the ordering
key to a single Object and making one chooseThread call would therefore move
ledgers with ids >= 2^31 to a different worker thread than today and than
the
other ledger-id keyed dispatches, so the two overload calls stay.
* Stub LedgerHandle.submitOrdered in ReadLastConfirmedAndEntryOpTest
The op now issues its speculative reads through the handle's executor
(LedgerHandle.submitOrdered), which on the Mockito-mocked handle of this
test returned null, so the speculative reads were never sent and the test
waited forever for them. Route the stubbed method to the test's ordered
scheduler, keyed by the ledger id as the op did before.
---
.../apache/bookkeeper/client/BatchedReadOp.java | 6 +-
.../apache/bookkeeper/client/ForceLedgerOp.java | 2 +-
.../apache/bookkeeper/client/LedgerCreateOp.java | 32 +++-
.../org/apache/bookkeeper/client/LedgerHandle.java | 50 ++++-
.../apache/bookkeeper/client/LedgerHandleAdv.java | 10 +-
.../org/apache/bookkeeper/client/LedgerOpenOp.java | 101 ++++++----
.../apache/bookkeeper/client/LedgerRecoveryOp.java | 1 +
.../org/apache/bookkeeper/client/PendingAddOp.java | 2 +-
.../apache/bookkeeper/client/PendingReadLacOp.java | 2 +-
.../apache/bookkeeper/client/PendingReadOp.java | 5 +-
.../bookkeeper/client/PendingWriteLacOp.java | 2 +-
.../client/ReadLastConfirmedAndEntryOp.java | 4 +-
.../bookkeeper/client/ReadLastConfirmedOp.java | 22 ++-
.../bookkeeper/client/ReadOnlyLedgerHandle.java | 14 +-
.../org/apache/bookkeeper/client/ReadOpBase.java | 4 +-
.../bookkeeper/client/TryReadLastConfirmedOp.java | 2 +-
.../bookkeeper/client/api/CreateAdvBuilder.java | 14 ++
.../bookkeeper/client/api/CreateBuilder.java | 18 ++
.../apache/bookkeeper/client/api/OpenBuilder.java | 18 ++
.../bookkeeper/client/impl/OpenBuilderBase.java | 7 +
.../org/apache/bookkeeper/proto/AddCompletion.java | 12 +-
.../bookkeeper/proto/BatchedReadCompletion.java | 6 +-
.../org/apache/bookkeeper/proto/BookieClient.java | 117 +++++++++++-
.../apache/bookkeeper/proto/BookieClientImpl.java | 75 +++++---
.../apache/bookkeeper/proto/CompletionValue.java | 21 ++-
.../bookkeeper/proto/ForceLedgerCompletion.java | 6 +-
.../bookkeeper/proto/PerChannelBookieClient.java | 66 ++++---
.../apache/bookkeeper/proto/ReadCompletion.java | 6 +-
.../apache/bookkeeper/proto/ReadLacCompletion.java | 6 +-
.../bookkeeper/proto/WriteLacCompletion.java | 6 +-
.../bookkeeper/client/MockBookKeeperTestCase.java | 18 ++
.../bookkeeper/client/PendingWriteLacOpTest.java | 3 +-
.../client/ReadLastConfirmedAndEntryOpTest.java | 4 +
.../api/BookKeeperBuildersOpenLedgerTest.java | 5 +-
.../bookkeeper/client/api/OrderingKeyTest.java | 209 +++++++++++++++++++++
.../apache/bookkeeper/proto/MockBookieClient.java | 36 ++--
.../proto/TestPerChannelBookieClient.java | 2 +-
.../apache/bookkeeper/test/BookieClientTest.java | 75 ++++++++
38 files changed, 836 insertions(+), 153 deletions(-)
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BatchedReadOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BatchedReadOp.java
index 89a976644e..6866abd8c9 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BatchedReadOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BatchedReadOp.java
@@ -137,10 +137,12 @@ public class BatchedReadOp extends ReadOpBase implements
BatchedReadEntryCallbac
if (isRecoveryRead) {
int flags = BookieProtocol.FLAG_HIGH_PRIORITY |
BookieProtocol.FLAG_DO_FENCING;
clientCtx.getBookieClient().batchReadEntries(to, lh.ledgerId,
entry.eId,
- maxCount, maxSize, this, new ReadContext(bookieIndex, to,
entry), flags, lh.ledgerKey);
+ maxCount, maxSize, this, new ReadContext(bookieIndex, to,
entry), flags, lh.ledgerKey, false,
+ lh.executor);
} else {
clientCtx.getBookieClient().batchReadEntries(to, lh.ledgerId,
entry.eId, maxCount, maxSize,
- this, new ReadContext(bookieIndex, to, entry),
BookieProtocol.FLAG_NONE);
+ this, new ReadContext(bookieIndex, to, entry),
BookieProtocol.FLAG_NONE, null, false,
+ lh.executor);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ForceLedgerOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ForceLedgerOp.java
index c0735e9758..93169b1e31 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ForceLedgerOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ForceLedgerOp.java
@@ -55,7 +55,7 @@ class ForceLedgerOp implements Runnable, ForceLedgerCallback {
}
void sendForceLedgerRequest(int bookieIndex) {
- bookieClient.forceLedger(currentEnsemble.get(bookieIndex),
lh.ledgerId, this, bookieIndex);
+ bookieClient.forceLedger(currentEnsemble.get(bookieIndex),
lh.ledgerId, this, bookieIndex, lh.executor);
}
@Override
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
index 69679ee9c1..ee0be88205 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
@@ -75,6 +75,7 @@ class LedgerCreateOp {
final OpStatsLogger createOpLogger;
final BookKeeperClientStats clientStats;
final Logger parentLogger;
+ final Object orderingKey;
boolean adv = false;
boolean generateLedgerId = true;
@@ -107,15 +108,20 @@ class LedgerCreateOp {
EnumSet<WriteFlag> writeFlags,
BookKeeperClientStats clientStats) {
this(bk, ensembleSize, writeQuorumSize, ackQuorumSize, digestType,
passwd, cb, ctx,
- customMetadata, writeFlags, clientStats, null);
+ customMetadata, writeFlags, clientStats, null, null);
}
+ /**
+ * @param orderingKey key selecting the worker thread that runs every
callback of the created handle;
+ * {@code null} selects it by ledger id
+ */
LedgerCreateOp(
BookKeeper bk, int ensembleSize, int writeQuorumSize, int
ackQuorumSize, DigestType digestType,
byte[] passwd, CreateCallback cb, Object ctx, final Map<String,
byte[]> customMetadata,
EnumSet<WriteFlag> writeFlags,
BookKeeperClientStats clientStats,
- Logger parentLogger) {
+ Logger parentLogger,
+ Object orderingKey) {
this.bk = bk;
this.metadataFormatVersion =
bk.getConf().getLedgerMetadataFormatVersion();
this.ensembleSize = ensembleSize;
@@ -131,6 +137,7 @@ class LedgerCreateOp {
this.createOpLogger = clientStats.getCreateOpLogger();
this.clientStats = clientStats;
this.parentLogger = parentLogger;
+ this.orderingKey = orderingKey;
}
/**
@@ -255,10 +262,10 @@ class LedgerCreateOp {
try {
if (adv) {
lh = new LedgerHandleAdv(bk.getClientCtx(), ledgerId,
writtenMetadata,
- digestType, passwd, writeFlags,
parentLogger);
+ digestType, passwd, writeFlags,
parentLogger, orderingKey);
} else {
lh = new LedgerHandle(bk.getClientCtx(), ledgerId,
writtenMetadata, digestType, passwd, writeFlags,
- parentLogger);
+ parentLogger, orderingKey);
}
} catch (GeneralSecurityException e) {
log.error()
@@ -317,6 +324,7 @@ class LedgerCreateOp {
org.apache.bookkeeper.client.api.DigestType.CRC32;
private Map<String, byte[]> builderCustomMetadata =
Collections.emptyMap();
private Logger builderParentLogger;
+ private Object builderOrderingKey;
CreateBuilderImpl(BookKeeper bk) {
this.bk = bk;
@@ -371,6 +379,12 @@ class LedgerCreateOp {
return this;
}
+ @Override
+ public CreateBuilder withOrderingKey(Object orderingKey) {
+ this.builderOrderingKey = orderingKey;
+ return this;
+ }
+
@Override
public CreateAdvBuilder makeAdv() {
return new CreateAdvBuilderImpl(this);
@@ -437,7 +451,7 @@ class LedgerCreateOp {
LedgerCreateOp op = new LedgerCreateOp(bk, builderEnsembleSize,
builderWriteQuorumSize, builderAckQuorumSize,
DigestType.fromApiDigestType(builderDigestType),
builderPassword, cb, null, builderCustomMetadata,
builderWriteFlags,
- bk.getClientCtx().getClientStats(), builderParentLogger);
+ bk.getClientCtx().getClientStats(), builderParentLogger,
builderOrderingKey);
ReentrantReadWriteLock closeLock = bk.getCloseLock();
closeLock.readLock().lock();
try {
@@ -467,6 +481,12 @@ class LedgerCreateOp {
return this;
}
+ @Override
+ public CreateAdvBuilder withOrderingKey(Object orderingKey) {
+ parent.builderOrderingKey = orderingKey;
+ return this;
+ }
+
@Override
public CompletableFuture<WriteAdvHandle> execute() {
CompletableFuture<WriteAdvHandle> future = new
CompletableFuture<>();
@@ -499,7 +519,7 @@ class LedgerCreateOp {
parent.builderPassword, cb, null,
parent.builderCustomMetadata,
parent.builderWriteFlags,
parent.bk.getClientCtx().getClientStats(),
- parent.builderParentLogger);
+ parent.builderParentLogger, parent.builderOrderingKey);
ReentrantReadWriteLock closeLock = parent.bk.getCloseLock();
closeLock.readLock().lock();
try {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index 0a1005f6ce..a4699a15b7 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -30,7 +30,9 @@ import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.RateLimiter;
+import com.google.common.util.concurrent.SettableFuture;
import io.github.merlimat.slog.Logger;
import io.github.merlimat.slog.LoggerBuilder;
import io.netty.buffer.ByteBuf;
@@ -46,6 +48,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
@@ -190,6 +193,20 @@ public class LedgerHandle implements WriteHandle {
EnumSet<WriteFlag> writeFlags,
Logger parentLogger)
throws GeneralSecurityException, NumberFormatException {
+ this(clientCtx, ledgerId, versionedMetadata, digestType, password,
writeFlags, parentLogger, null);
+ }
+
+ /**
+ * @param orderingKey key selecting the worker thread that runs every
callback of this handle;
+ * {@code null} selects it by ledger id
+ */
+ LedgerHandle(ClientContext clientCtx,
+ long ledgerId, Versioned<LedgerMetadata> versionedMetadata,
+ BookKeeper.DigestType digestType, byte[] password,
+ EnumSet<WriteFlag> writeFlags,
+ Logger parentLogger,
+ Object orderingKey)
+ throws GeneralSecurityException, NumberFormatException {
LoggerBuilder builder = Logger.get(LedgerHandle.class).with();
if (parentLogger != null) {
builder = builder.ctx(parentLogger);
@@ -213,7 +230,12 @@ public class LedgerHandle implements WriteHandle {
this.pendingAddsSequenceHead = lastAddConfirmed;
this.ledgerId = ledgerId;
- this.executor = clientCtx.getMainWorkerPool().chooseThread(ledgerId);
+ // Two calls on purpose: chooseThread(long) hashes the raw id while
chooseThread(Object) goes through
+ // hashCode(), and Long.hashCode folds the high bits. Boxing the id
would move ledgers with ids >= 2^31
+ // to a different thread than the other ledger-id keyed dispatches
(e.g. OrderedGenericCallback).
+ this.executor = orderingKey == null
+ ? clientCtx.getMainWorkerPool().chooseThread(ledgerId)
+ : clientCtx.getMainWorkerPool().chooseThread(orderingKey);
if (clientCtx.getConf().enableStickyReads
&& getLedgerMetadata().getEnsembleSize() ==
getLedgerMetadata().getWriteQuorumSize()) {
@@ -1001,7 +1023,7 @@ public class LedgerHandle implements WriteHandle {
batchReadEntriesInternalAsync(startEntry, maxCount, maxSize, false)
.whenCompleteAsync((entries, error) ->
completeBatchReadUnconfirmed(
startEntry, lastEntry, entries, error, future),
- clientCtx.getMainWorkerPool().chooseThread(ledgerId));
+ executor);
return future;
}
@@ -1175,7 +1197,7 @@ public class LedgerHandle implements WriteHandle {
cb.readComplete(Code.UnexpectedConditionException,
LedgerHandle.this, null, ctx);
}
}
- }, clientCtx.getMainWorkerPool().chooseThread(ledgerId));
+ }, executor);
} else {
cb.readComplete(Code.ClientClosedException, LedgerHandle.this,
null, ctx);
}
@@ -1209,7 +1231,7 @@ public class LedgerHandle implements WriteHandle {
cb.readComplete(Code.UnexpectedConditionException, LedgerHandle.this, null,
ctx);
}
}
- }, clientCtx.getMainWorkerPool().chooseThread(ledgerId));
+ }, executor);
} else {
cb.readComplete(Code.ClientClosedException, LedgerHandle.this,
null, ctx);
}
@@ -1815,6 +1837,7 @@ public class LedgerHandle implements WriteHandle {
ledgerId,
getCurrentEnsemble(),
ledgerKey,
+ executor,
innercb).initiate();
}
@@ -2424,7 +2447,7 @@ public class LedgerHandle implements WriteHandle {
unsetSuccessAndSendWriteRequest(newEnsemble,
replaced);
}
}
- }, clientCtx.getMainWorkerPool().chooseThread(ledgerId));
+ }, executor);
}
void unsetSuccessAndSendWriteRequest(List<BookieId> ensemble, final
Set<Integer> bookies) {
@@ -2512,6 +2535,23 @@ public class LedgerHandle implements WriteHandle {
executor.execute(runnable);
}
+ /**
+ * Run the task in the thread pinned to the ledger, exposing its result as
a future.
+ * @param task
+ * @throws RejectedExecutionException
+ */
+ <T> ListenableFuture<T> submitOrdered(Callable<T> task) throws
RejectedExecutionException {
+ SettableFuture<T> future = SettableFuture.create();
+ executeOrdered(() -> {
+ try {
+ future.set(task.call());
+ } catch (Throwable t) {
+ future.setException(t);
+ }
+ });
+ return future;
+ }
+
@VisibleForTesting
public Queue<PendingAddOp> getPendingAddOps() {
return pendingAddOps;
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
index 31be642d81..92146b8230 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
@@ -65,7 +65,15 @@ public class LedgerHandleAdv extends LedgerHandle implements
WriteAdvHandle {
BookKeeper.DigestType digestType, byte[] password,
EnumSet<WriteFlag> writeFlags,
Logger parentLogger)
throws GeneralSecurityException, NumberFormatException {
- super(clientCtx, ledgerId, metadata, digestType, password, writeFlags,
parentLogger);
+ this(clientCtx, ledgerId, metadata, digestType, password, writeFlags,
parentLogger, null);
+ }
+
+ LedgerHandleAdv(ClientContext clientCtx,
+ long ledgerId, Versioned<LedgerMetadata> metadata,
+ BookKeeper.DigestType digestType, byte[] password,
EnumSet<WriteFlag> writeFlags,
+ Logger parentLogger, Object orderingKey)
+ throws GeneralSecurityException, NumberFormatException {
+ super(clientCtx, ledgerId, metadata, digestType, password, writeFlags,
parentLogger, orderingKey);
pendingAddOps = new PriorityBlockingQueue<PendingAddOp>(10, new
PendingOpsComparator());
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
index 4bae409ec7..79a786a361 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
@@ -28,6 +28,7 @@ import io.github.merlimat.slog.LoggerBuilder;
import java.security.GeneralSecurityException;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.bookkeeper.client.AsyncCallback.OpenCallback;
@@ -39,6 +40,7 @@ import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.impl.OpenBuilderBase;
import org.apache.bookkeeper.common.util.MathUtils;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.util.OrderedGenericCallback;
import org.apache.bookkeeper.versioning.Versioned;
@@ -77,6 +79,7 @@ class LedgerOpenOp {
final DigestType suggestedDigestType;
final boolean enableDigestAutodetection;
final Logger parentLogger;
+ final Object orderingKey;
/**
* Constructor.
@@ -91,12 +94,16 @@ class LedgerOpenOp {
public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
long ledgerId, DigestType digestType, byte[] passwd,
OpenCallback cb, Object ctx) {
- this(bk, clientStats, ledgerId, digestType, passwd, cb, ctx, null);
+ this(bk, clientStats, ledgerId, digestType, passwd, cb, ctx, null,
null);
}
+ /**
+ * @param orderingKey key selecting the worker thread that runs every
callback of the opened handle;
+ * {@code null} selects it by ledger id
+ */
public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
long ledgerId, DigestType digestType, byte[] passwd,
- OpenCallback cb, Object ctx, Logger parentLogger) {
+ OpenCallback cb, Object ctx, Logger parentLogger,
Object orderingKey) {
LoggerBuilder builder = Logger.get(LedgerOpenOp.class).with();
if (parentLogger != null) {
builder = builder.ctx(parentLogger);
@@ -111,6 +118,7 @@ class LedgerOpenOp {
this.suggestedDigestType = digestType;
this.openOpLogger = clientStats.getOpenOpLogger();
this.parentLogger = parentLogger;
+ this.orderingKey = orderingKey;
}
public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
@@ -127,6 +135,7 @@ class LedgerOpenOp {
this.suggestedDigestType = bk.conf.getBookieRecoveryDigestType();
this.openOpLogger = clientStats.getOpenOpLogger();
this.parentLogger = null;
+ this.orderingKey = null;
}
/**
@@ -139,7 +148,8 @@ class LedgerOpenOp {
* Asynchronously read the ledger metadata node.
*/
bk.getLedgerManager().readLedgerMetadata(ledgerId)
- .thenAcceptAsync(this::openWithMetadata,
bk.getScheduler().chooseThread(ledgerId))
+ .thenAcceptAsync(this::openWithMetadata, orderingKey == null
+ ? bk.getScheduler().chooseThread(ledgerId) :
bk.getScheduler().chooseThread(orderingKey))
.exceptionally(exception -> {
openComplete(BKException.getExceptionCode(exception),
null);
return null;
@@ -229,7 +239,7 @@ class LedgerOpenOp {
// Therefore, if a user needs to the feature that update metadata
automatically, he will set
// "keepUpdateMetadata" to "true",
lh = new ReadOnlyLedgerHandle(bk.getClientCtx(), ledgerId,
versionedMetadata, digestType,
- passwd, watchImmediately,
parentLogger);
+ passwd, watchImmediately,
parentLogger, orderingKey);
} catch (GeneralSecurityException e) {
log.error().exception(e).attr("ledgerId", ledgerId).log("Security
exception while opening ledger");
openComplete(BKException.Code.DigestNotInitializedException, null);
@@ -248,35 +258,7 @@ class LedgerOpenOp {
}
if (doRecovery) {
- lh.recover(new
OrderedGenericCallback<Void>(bk.getMainWorkerPool(), ledgerId) {
- @Override
- public void safeOperationComplete(int rc, Void result) {
- if (rc == BKException.Code.OK) {
- openComplete(BKException.Code.OK, lh);
- if (!watchImmediately && keepUpdateMetadata) {
- lh.registerLedgerMetadataListener();
- }
- } else {
- closeLedgerHandleAsync().whenComplete((ignore, ex) -> {
- if (ex != null) {
- log.error()
- .exception(ex)
- .log("Ledger close failed");
- }
- if (rc ==
BKException.Code.UnauthorizedAccessException
- || rc ==
BKException.Code.TimeoutException) {
- openComplete(bk.getReturnRc(rc), null);
- } else {
-
openComplete(bk.getReturnRc(BKException.Code.LedgerRecoveryException), null);
- }
- });
- }
- }
- @Override
- public String toString() {
- return String.format("Recover(%d)", ledgerId);
- }
- });
+ lh.recover(recoveryCallback(watchImmediately));
} else {
lh.asyncReadLastConfirmed(new ReadLastConfirmedCallback() {
@Override
@@ -310,6 +292,57 @@ class LedgerOpenOp {
}
}
+ /**
+ * Callback completing the open once recovery is done, run on the handle's
thread. Without an ordering key
+ * this is the ledger-id keyed {@link OrderedGenericCallback}; with one,
the completion is submitted to the
+ * handle's executor, which is the thread selected by that key.
+ */
+ private GenericCallback<Void> recoveryCallback(boolean watchImmediately) {
+ if (orderingKey == null) {
+ return new OrderedGenericCallback<Void>(bk.getMainWorkerPool(),
ledgerId) {
+ @Override
+ public void safeOperationComplete(int rc, Void result) {
+ recoveryComplete(rc, watchImmediately);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("Recover(%d)", ledgerId);
+ }
+ };
+ }
+ return (rc, result) -> {
+ try {
+ lh.executeOrdered(() -> recoveryComplete(rc,
watchImmediately));
+ } catch (RejectedExecutionException ree) {
+ log.warn().exception(ree).log("Failed to submit recovery
completion callback");
+ }
+ };
+ }
+
+ private void recoveryComplete(int rc, boolean watchImmediately) {
+ if (rc == BKException.Code.OK) {
+ openComplete(BKException.Code.OK, lh);
+ if (!watchImmediately && keepUpdateMetadata) {
+ lh.registerLedgerMetadataListener();
+ }
+ } else {
+ closeLedgerHandleAsync().whenComplete((ignore, ex) -> {
+ if (ex != null) {
+ log.error()
+ .exception(ex)
+ .log("Ledger close failed");
+ }
+ if (rc == BKException.Code.UnauthorizedAccessException
+ || rc == BKException.Code.TimeoutException) {
+ openComplete(bk.getReturnRc(rc), null);
+ } else {
+
openComplete(bk.getReturnRc(BKException.Code.LedgerRecoveryException), null);
+ }
+ });
+ }
+ }
+
void openComplete(int rc, LedgerHandle lh) {
if (BKException.Code.OK != rc) {
openOpLogger.registerFailedEvent(MathUtils.elapsedNanos(startTime),
TimeUnit.NANOSECONDS);
@@ -349,7 +382,7 @@ class LedgerOpenOp {
LedgerOpenOp op = new LedgerOpenOp(bk,
bk.getClientCtx().getClientStats(),
ledgerId,
fromApiDigestType(digestType),
- password, cb, null,
parentLogger);
+ password, cb, null,
parentLogger, orderingKey);
ReentrantReadWriteLock closeLock = bk.getCloseLock();
closeLock.readLock().lock();
try {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
index a019157038..76db2a969f 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
@@ -96,6 +96,7 @@ class LedgerRecoveryOp implements ReadEntryListener,
AddCallback {
lh.ledgerId,
lh.getCurrentEnsemble(),
lh.ledgerKey,
+ lh.executor,
new ReadLastConfirmedOp.LastConfirmedDataCallback() {
@Override
public void readLastConfirmedDataComplete(int rc,
RecoveryData data) {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
index e2cd81ef2f..be6ae59b41 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
@@ -148,7 +148,7 @@ class PendingAddOp implements WriteCallback {
clientCtx.getBookieClient().addEntry(ensemble.get(bookieIndex),
lh.ledgerId, lh.ledgerKey,
entryId, toSend, this, bookieIndex,
- flags, allowFailFast,
lh.writeFlags);
+ flags, allowFailFast,
lh.writeFlags, lh.executor);
++pendingWriteRequests;
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadLacOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadLacOp.java
index 1f7ec85013..c38417f05b 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadLacOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadLacOp.java
@@ -71,7 +71,7 @@ class PendingReadLacOp implements ReadLacCallback {
public void initiate() {
for (int i = 0; i < currentEnsemble.size(); i++) {
- bookieClient.readLac(currentEnsemble.get(i), lh.ledgerId, this, i);
+ bookieClient.readLac(currentEnsemble.get(i), lh.ledgerId, this, i,
lh.executor);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
index 10139e999c..fa441776dc 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java
@@ -181,10 +181,11 @@ class PendingReadOp extends ReadOpBase implements
ReadEntryCallback {
if (isRecoveryRead) {
int flags = BookieProtocol.FLAG_HIGH_PRIORITY |
BookieProtocol.FLAG_DO_FENCING;
clientCtx.getBookieClient().readEntry(to, lh.ledgerId, entry.eId,
- this, new ReadContext(bookieIndex, to, entry), flags,
lh.ledgerKey);
+ this, new ReadContext(bookieIndex, to, entry), flags,
lh.ledgerKey, false, lh.executor);
} else {
clientCtx.getBookieClient().readEntry(to, lh.ledgerId, entry.eId,
- this, new ReadContext(bookieIndex, to, entry),
BookieProtocol.FLAG_NONE);
+ this, new ReadContext(bookieIndex, to, entry),
BookieProtocol.FLAG_NONE, null, false,
+ lh.executor);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingWriteLacOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingWriteLacOp.java
index 373c6832ac..62ddfe4ce5 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingWriteLacOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingWriteLacOp.java
@@ -73,7 +73,7 @@ class PendingWriteLacOp implements WriteLacCallback {
void sendWriteLacRequest(int bookieIndex, ByteBufList toSend) {
clientCtx.getBookieClient().writeLac(currentEnsemble.get(bookieIndex),
- lh.ledgerId, lh.ledgerKey, lac,
toSend, this, bookieIndex);
+ lh.ledgerId, lh.ledgerKey, lac,
toSend, this, bookieIndex, lh.executor);
}
void initiate(ByteBufList toSend) {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOp.java
index 99a01a8078..e47df2146c 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOp.java
@@ -505,7 +505,7 @@ class ReadLastConfirmedAndEntryOp implements
BookkeeperInternalCallbacks.ReadEnt
*/
@Override
public ListenableFuture<Boolean> issueSpeculativeRequest() {
- return clientCtx.getMainWorkerPool().submitOrdered(lh.getId(), new
Callable<Boolean>() {
+ return lh.submitOrdered(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
if (!requestComplete.get() && !request.isComplete()
@@ -554,7 +554,7 @@ class ReadLastConfirmedAndEntryOp implements
BookkeeperInternalCallbacks.ReadEnt
prevEntryId,
timeOutInMillis,
true,
- this, new ReadLastConfirmedAndEntryContext(bookieIndex, to));
+ this, new ReadLastConfirmedAndEntryContext(bookieIndex, to),
lh.executor);
this.numResponsesPending++;
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedOp.java
index 83c3fcc62f..394529b852 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadLastConfirmedOp.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.client;
import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import java.util.List;
+import java.util.concurrent.Executor;
import lombok.CustomLog;
import org.apache.bookkeeper.client.BKException.BKDigestMatchException;
import org.apache.bookkeeper.net.BookieId;
@@ -39,6 +40,7 @@ class ReadLastConfirmedOp implements ReadEntryCallback {
private final byte[] ledgerKey;
private final BookieClient bookieClient;
private final DigestManager digestManager;
+ private final Executor callbackExecutor;
private int numResponsesPending;
private RecoveryData maxRecoveredData;
private volatile boolean completed = false;
@@ -62,6 +64,21 @@ class ReadLastConfirmedOp implements ReadEntryCallback {
List<BookieId> ensemble,
byte[] ledgerKey,
LastConfirmedDataCallback cb) {
+ this(bookieClient, schedule, digestManager, ledgerId, ensemble,
ledgerKey, null, cb);
+ }
+
+ /**
+ * @param callbackExecutor executor running the read callbacks; {@code
null} uses the worker thread
+ * selected by ledger id
+ */
+ public ReadLastConfirmedOp(BookieClient bookieClient,
+ DistributionSchedule schedule,
+ DigestManager digestManager,
+ long ledgerId,
+ List<BookieId> ensemble,
+ byte[] ledgerKey,
+ Executor callbackExecutor,
+ LastConfirmedDataCallback cb) {
this.cb = cb;
this.bookieClient = bookieClient;
this.maxRecoveredData = new
RecoveryData(LedgerHandle.INVALID_ENTRY_ID, 0);
@@ -71,6 +88,7 @@ class ReadLastConfirmedOp implements ReadEntryCallback {
this.ledgerId = ledgerId;
this.ledgerKey = ledgerKey;
this.digestManager = digestManager;
+ this.callbackExecutor = callbackExecutor;
}
public void initiate() {
@@ -78,7 +96,7 @@ class ReadLastConfirmedOp implements ReadEntryCallback {
bookieClient.readEntry(currentEnsemble.get(i),
ledgerId,
BookieProtocol.LAST_ADD_CONFIRMED,
- this, i, BookieProtocol.FLAG_NONE);
+ this, i, BookieProtocol.FLAG_NONE, null,
false, callbackExecutor);
}
}
@@ -88,7 +106,7 @@ class ReadLastConfirmedOp implements ReadEntryCallback {
ledgerId,
BookieProtocol.LAST_ADD_CONFIRMED,
this, i, BookieProtocol.FLAG_DO_FENCING,
- ledgerKey);
+ ledgerKey, false, callbackExecutor);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
index bdd9bb98c1..f0933c947e 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
@@ -105,7 +105,17 @@ class ReadOnlyLedgerHandle extends LedgerHandle implements
LedgerMetadataListene
boolean watchImmediately,
Logger parentLogger)
throws GeneralSecurityException, NumberFormatException {
- super(clientCtx, ledgerId, metadata, digestType, password,
WriteFlag.NONE, parentLogger);
+ this(clientCtx, ledgerId, metadata, digestType, password,
watchImmediately, parentLogger, null);
+ }
+
+ ReadOnlyLedgerHandle(ClientContext clientCtx,
+ long ledgerId, Versioned<LedgerMetadata> metadata,
+ BookKeeper.DigestType digestType, byte[] password,
+ boolean watchImmediately,
+ Logger parentLogger,
+ Object orderingKey)
+ throws GeneralSecurityException, NumberFormatException {
+ super(clientCtx, ledgerId, metadata, digestType, password,
WriteFlag.NONE, parentLogger, orderingKey);
if (watchImmediately) {
registerLedgerMetadataListener();
}
@@ -177,7 +187,7 @@ class ReadOnlyLedgerHandle extends LedgerHandle implements
LedgerMetadataListene
if (Version.Occurred.BEFORE == occurred) { // the metadata is updated
try {
- clientCtx.getMainWorkerPool().executeOrdered(ledgerId, new
MetadataUpdater(newMetadata));
+ executeOrdered(new MetadataUpdater(newMetadata));
} catch (RejectedExecutionException ree) {
log.error()
.attr("newMetadata", newMetadata)
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOpBase.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOpBase.java
index 00f32ecee5..1ec0589bbb 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOpBase.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOpBase.java
@@ -95,7 +95,7 @@ public abstract class ReadOpBase implements Runnable {
}
public void submit() {
- clientCtx.getMainWorkerPool().executeOrdered(lh.ledgerId, this);
+ lh.executeOrdered(this);
}
@Override
@@ -256,7 +256,7 @@ public abstract class ReadOpBase implements Runnable {
*/
@Override
public ListenableFuture<Boolean> issueSpeculativeRequest() {
- return clientCtx.getMainWorkerPool().submitOrdered(lh.getId(), new
Callable<Boolean>() {
+ return lh.submitOrdered(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
if (!isComplete() && null !=
maybeSendSpeculativeRead(heardFromHostsBitSet)) {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TryReadLastConfirmedOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TryReadLastConfirmedOp.java
index 926ddb757b..2b760296e0 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TryReadLastConfirmedOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TryReadLastConfirmedOp.java
@@ -59,7 +59,7 @@ class TryReadLastConfirmedOp implements ReadEntryCallback {
bookieClient.readEntry(currentEnsemble.get(i),
lh.ledgerId,
BookieProtocol.LAST_ADD_CONFIRMED,
- this, i, BookieProtocol.FLAG_NONE);
+ this, i, BookieProtocol.FLAG_NONE, null,
false, lh.executor);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateAdvBuilder.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateAdvBuilder.java
index 9a316b1fb0..df1d73d3da 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateAdvBuilder.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateAdvBuilder.java
@@ -42,4 +42,18 @@ public interface CreateAdvBuilder extends
OpBuilder<WriteAdvHandle> {
* @return the builder itself
*/
CreateAdvBuilder withLedgerId(long ledgerId);
+
+ /**
+ * Set the key used to select the client worker thread on which every
callback of the resulting
+ * {@link WriteAdvHandle} runs. By default the thread is selected by
ledger id.
+ *
+ * @param orderingKey the ordering key; {@code null} (the default) selects
the thread by ledger id
+ *
+ * @return the builder itself
+ *
+ * @see CreateBuilder#withOrderingKey(Object)
+ */
+ default CreateAdvBuilder withOrderingKey(Object orderingKey) {
+ return this;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
index af10a10eae..181c1b3170 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
@@ -141,4 +141,22 @@ public interface CreateBuilder extends
OpBuilder<WriteHandle> {
return this;
}
+ /**
+ * Set the key used to select the client worker thread on which every
callback of the resulting
+ * {@link WriteHandle} runs (add, read and close completions, as well as
the completion of this create
+ * operation). By default the thread is selected by ledger id.
+ *
+ * <p>The thread is resolved with {@link
org.apache.bookkeeper.common.util.OrderedExecutor#chooseThread(Object)}
+ * on the client's main worker pool, so an application that already runs
its own per-entity work on
+ * {@code bookKeeper.getMainWorkerPool().chooseThread(key)} can pass the
same key here and have the ledger's
+ * callbacks delivered on that very thread, avoiding a cross-thread hop
per completion.
+ *
+ * @param orderingKey the ordering key; {@code null} (the default) selects
the thread by ledger id
+ *
+ * @return the builder itself
+ */
+ default CreateBuilder withOrderingKey(Object orderingKey) {
+ return this;
+ }
+
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/OpenBuilder.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/OpenBuilder.java
index b0951b9972..67888ccf40 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/OpenBuilder.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/OpenBuilder.java
@@ -107,4 +107,22 @@ public interface OpenBuilder extends OpBuilder<ReadHandle>
{
return this;
}
+ /**
+ * Set the key used to select the client worker thread on which every
callback of the resulting
+ * {@link ReadHandle} runs (read completions, recovery and the completion
of this open operation).
+ * By default the thread is selected by ledger id.
+ *
+ * <p>The thread is resolved with {@link
org.apache.bookkeeper.common.util.OrderedExecutor#chooseThread(Object)}
+ * on the client's main worker pool, so an application that already runs
its own per-entity work on
+ * {@code bookKeeper.getMainWorkerPool().chooseThread(key)} can pass the
same key here and have the ledger's
+ * callbacks delivered on that very thread, avoiding a cross-thread hop
per completion.
+ *
+ * @param orderingKey the ordering key; {@code null} (the default) selects
the thread by ledger id
+ *
+ * @return the builder itself
+ */
+ default OpenBuilder withOrderingKey(Object orderingKey) {
+ return this;
+ }
+
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/OpenBuilderBase.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/OpenBuilderBase.java
index 45afcce91e..262dfe7d76 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/OpenBuilderBase.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/OpenBuilderBase.java
@@ -38,6 +38,7 @@ public abstract class OpenBuilderBase implements OpenBuilder {
protected DigestType digestType = DigestType.CRC32;
protected boolean keepUpdateMetadata = false;
protected Logger parentLogger;
+ protected Object orderingKey;
@Override
public OpenBuilder withLedgerId(long ledgerId) {
@@ -75,6 +76,12 @@ public abstract class OpenBuilderBase implements OpenBuilder
{
return this;
}
+ @Override
+ public OpenBuilder withOrderingKey(Object orderingKey) {
+ this.orderingKey = orderingKey;
+ return this;
+ }
+
protected int validate() {
if (ledgerId < 0) {
log.error().attr("ledgerId", ledgerId).log("invalid ledgerId < 0");
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AddCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AddCompletion.java
index 1ed31b8dd0..d7db0d8f35 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AddCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AddCompletion.java
@@ -22,6 +22,7 @@
package org.apache.bookkeeper.proto;
import io.netty.util.Recycler;
+import java.util.concurrent.Executor;
import lombok.CustomLog;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.common.util.MathUtils;
@@ -35,9 +36,11 @@ class AddCompletion extends CompletionValue implements
BookkeeperInternalCallbac
final
BookkeeperInternalCallbacks.WriteCallback originalCallback,
final Object originalCtx,
final long ledgerId, final long
entryId,
- PerChannelBookieClient
perChannelBookieClient) {
+ PerChannelBookieClient
perChannelBookieClient,
+ Executor callbackExecutor) {
AddCompletion completion = ADD_COMPLETION_RECYCLER.get();
- completion.reset(key, originalCallback, originalCtx, ledgerId,
entryId, perChannelBookieClient);
+ completion.reset(key, originalCallback, originalCtx, ledgerId,
entryId, perChannelBookieClient,
+ callbackExecutor);
return completion;
}
@@ -55,7 +58,8 @@ class AddCompletion extends CompletionValue implements
BookkeeperInternalCallbac
final BookkeeperInternalCallbacks.WriteCallback
originalCallback,
final Object originalCtx,
final long ledgerId, final long entryId,
- PerChannelBookieClient perChannelBookieClient) {
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
this.key = key;
this.originalCallback = originalCallback;
this.ctx = originalCtx;
@@ -66,6 +70,7 @@ class AddCompletion extends CompletionValue implements
BookkeeperInternalCallbac
this.opLogger = perChannelBookieClient.addEntryOpLogger;
this.timeoutOpLogger = perChannelBookieClient.addTimeoutOpLogger;
this.perChannelBookieClient = perChannelBookieClient;
+ this.callbackExecutor = callbackExecutor;
this.mdcContextMap =
perChannelBookieClient.preserveMdcForTaskExecution ? MDC.getCopyOfContextMap()
: null;
}
@@ -75,6 +80,7 @@ class AddCompletion extends CompletionValue implements
BookkeeperInternalCallbac
this.opLogger = null;
this.timeoutOpLogger = null;
this.perChannelBookieClient = null;
+ this.callbackExecutor = null;
this.mdcContextMap = null;
handle.recycle(this);
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BatchedReadCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BatchedReadCompletion.java
index f6e0366be8..5dc074b625 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BatchedReadCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BatchedReadCompletion.java
@@ -23,6 +23,7 @@ package org.apache.bookkeeper.proto;
import static org.apache.bookkeeper.client.LedgerHandle.INVALID_ENTRY_ID;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.util.ByteBufList;
@@ -34,8 +35,9 @@ class BatchedReadCompletion extends CompletionValue {
final
BookkeeperInternalCallbacks.BatchedReadEntryCallback originalCallback,
final Object originalCtx,
long ledgerId, final long entryId,
- PerChannelBookieClient
perChannelBookieClient) {
- super("BatchedRead", originalCtx, ledgerId, entryId,
perChannelBookieClient);
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
+ super("BatchedRead", originalCtx, ledgerId, entryId,
perChannelBookieClient, callbackExecutor);
this.opLogger = perChannelBookieClient.readEntryOpLogger;
this.timeoutOpLogger = perChannelBookieClient.readTimeoutOpLogger;
this.cb = (rc, ledgerId1, startEntryId, bufList, ctx) -> {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
index 8903cb37bc..41077112c2 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
@@ -24,6 +24,7 @@ import io.netty.util.ReferenceCounted;
import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.net.BookieId;
import
org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.BatchedReadEntryCallback;
@@ -38,6 +39,12 @@ import org.apache.bookkeeper.util.ByteBufList;
/**
* Low level client for talking to bookies.
+ *
+ * <p>The callback of a ledger operation runs on the client worker thread
selected by the ledger id, unless
+ * the caller passes a {@code callbackExecutor}: then every completion of that
request (response, failure and
+ * timeout) is submitted to that executor instead, which lets a ledger handle
keep all of its callbacks on the
+ * single thread it was assigned. The overloads without {@code
callbackExecutor} are equivalent to passing
+ * {@code null}.
*/
public interface BookieClient {
long PENDINGREQ_NOTWRITABLE_MASK = 0x01L << 62;
@@ -95,8 +102,20 @@ public interface BookieClient {
* @param cb the callback notified when the request completes
* @param ctx a context object passed to the callback on completion
*/
+ default void forceLedger(BookieId address, long ledgerId,
+ ForceLedgerCallback cb, Object ctx) {
+ forceLedger(address, ledgerId, cb, ctx, null);
+ }
+
+ /**
+ * Send a force request to the server, running the callback on {@code
callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #forceLedger(BookieId, long, ForceLedgerCallback, Object)
+ */
void forceLedger(BookieId address, long ledgerId,
- ForceLedgerCallback cb, Object ctx);
+ ForceLedgerCallback cb, Object ctx, Executor
callbackExecutor);
/**
* Read the last add confirmed for ledger {@code ledgerId} from the bookie
at
@@ -107,7 +126,18 @@ public interface BookieClient {
* @param cb the callback notified when the request completes
* @param ctx a context object passed to the callback on completion
*/
- void readLac(BookieId address, long ledgerId, ReadLacCallback cb, Object
ctx);
+ default void readLac(BookieId address, long ledgerId, ReadLacCallback cb,
Object ctx) {
+ readLac(address, ledgerId, cb, ctx, null);
+ }
+
+ /**
+ * Read the last add confirmed, running the callback on {@code
callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #readLac(BookieId, long, ReadLacCallback, Object)
+ */
+ void readLac(BookieId address, long ledgerId, ReadLacCallback cb, Object
ctx, Executor callbackExecutor);
/**
* Explicitly write the last add confirmed for ledger {@code ledgerId} to
the bookie at
@@ -121,8 +151,20 @@ public interface BookieClient {
* @param cb the callback notified when the request completes
* @param ctx a context object passed to the callback on completion
*/
+ default void writeLac(BookieId address, long ledgerId, byte[] masterKey,
+ long lac, ByteBufList toSend, WriteLacCallback cb,
Object ctx) {
+ writeLac(address, ledgerId, masterKey, lac, toSend, cb, ctx, null);
+ }
+
+ /**
+ * Explicitly write the last add confirmed, running the callback on {@code
callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #writeLac(BookieId, long, byte[], long, ByteBufList,
WriteLacCallback, Object)
+ */
void writeLac(BookieId address, long ledgerId, byte[] masterKey,
- long lac, ByteBufList toSend, WriteLacCallback cb, Object
ctx);
+ long lac, ByteBufList toSend, WriteLacCallback cb, Object
ctx, Executor callbackExecutor);
/**
* Add an entry for ledger {@code ledgerId} on the bookie at address
{@code address}.
@@ -140,9 +182,23 @@ public interface BookieClient {
* @param writeFlags a set of write flags
* {@link org.apache.bookkeeper.client.api.WriteFlag}
*/
+ default void addEntry(BookieId address, long ledgerId, byte[] masterKey,
+ long entryId, ReferenceCounted toSend, WriteCallback
cb, Object ctx,
+ int options, boolean allowFastFail,
EnumSet<WriteFlag> writeFlags) {
+ addEntry(address, ledgerId, masterKey, entryId, toSend, cb, ctx,
options, allowFastFail, writeFlags, null);
+ }
+
+ /**
+ * Add an entry, running the callback on {@code callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #addEntry(BookieId, long, byte[], long, ReferenceCounted,
WriteCallback, Object, int, boolean, EnumSet)
+ */
void addEntry(BookieId address, long ledgerId, byte[] masterKey,
long entryId, ReferenceCounted toSend, WriteCallback cb,
Object ctx,
- int options, boolean allowFastFail, EnumSet<WriteFlag>
writeFlags);
+ int options, boolean allowFastFail, EnumSet<WriteFlag>
writeFlags,
+ Executor callbackExecutor);
/**
* Read entry with a null masterkey, disallowing failfast.
@@ -177,9 +233,22 @@ public interface BookieClient {
* @param allowFastFail fail the read immediately if the channel is
non-writable
* {@link #isWritable(BookieId,long)}
*/
+ default void readEntry(BookieId address, long ledgerId, long entryId,
+ ReadEntryCallback cb, Object ctx, int flags, byte[]
masterKey,
+ boolean allowFastFail) {
+ readEntry(address, ledgerId, entryId, cb, ctx, flags, masterKey,
allowFastFail, null);
+ }
+
+ /**
+ * Read an entry, running the callback on {@code callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #readEntry(BookieId, long, long, ReadEntryCallback, Object, int,
byte[], boolean)
+ */
void readEntry(BookieId address, long ledgerId, long entryId,
ReadEntryCallback cb, Object ctx, int flags, byte[]
masterKey,
- boolean allowFastFail);
+ boolean allowFastFail, Executor callbackExecutor);
/**
* Batch read entries with a null masterkey, disallowing failfast.
@@ -218,9 +287,23 @@ public interface BookieClient {
* @param allowFastFail fail the read immediately if the channel is
non-writable
* {@link #isWritable(BookieId,long)}
*/
+ default void batchReadEntries(BookieId address, long ledgerId, long
startEntryId,
+ int maxCount, long maxSize, BatchedReadEntryCallback cb, Object
ctx,
+ int flags, byte[] masterKey, boolean allowFastFail) {
+ batchReadEntries(address, ledgerId, startEntryId, maxCount, maxSize,
cb, ctx, flags, masterKey,
+ allowFastFail, null);
+ }
+
+ /**
+ * Batch read entries, running the callback on {@code callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #batchReadEntries(BookieId, long, long, int, long,
BatchedReadEntryCallback, Object, int, byte[], boolean)
+ */
void batchReadEntries(BookieId address, long ledgerId, long startEntryId,
int maxCount, long maxSize, BatchedReadEntryCallback cb, Object
ctx,
- int flags, byte[] masterKey, boolean allowFastFail);
+ int flags, byte[] masterKey, boolean allowFastFail, Executor
callbackExecutor);
/**
* Send a long poll request to bookie, waiting for the last add confirmed
@@ -236,6 +319,25 @@ public interface BookieClient {
* @param cb the callback notified when the request completes
* @param ctx a context object passed to the callback on completion
*/
+ default void readEntryWaitForLACUpdate(BookieId address,
+ long ledgerId,
+ long entryId,
+ long previousLAC,
+ long timeOutInMillis,
+ boolean piggyBackEntry,
+ ReadEntryCallback cb,
+ Object ctx) {
+ readEntryWaitForLACUpdate(address, ledgerId, entryId, previousLAC,
timeOutInMillis, piggyBackEntry, cb, ctx,
+ null);
+ }
+
+ /**
+ * Send a long poll request to bookie, running the callback on {@code
callbackExecutor}.
+ *
+ * @param callbackExecutor executor on which the callback is run; {@code
null} runs it on the client
+ * worker thread selected by {@code ledgerId}
+ * @see #readEntryWaitForLACUpdate(BookieId, long, long, long, long,
boolean, ReadEntryCallback, Object)
+ */
void readEntryWaitForLACUpdate(BookieId address,
long ledgerId,
long entryId,
@@ -243,7 +345,8 @@ public interface BookieClient {
long timeOutInMillis,
boolean piggyBackEntry,
ReadEntryCallback cb,
- Object ctx);
+ Object ctx,
+ Executor callbackExecutor);
/**
* Read information about the bookie, from the bookie.
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClientImpl.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClientImpl.java
index 668c481c13..7ecafeb5a7 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClientImpl.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClientImpl.java
@@ -39,6 +39,7 @@ import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -132,6 +133,14 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
}
}
+ /**
+ * Runs {@code r} on the thread owning the callbacks of the ledger: the
caller-supplied executor when
+ * present, otherwise the worker thread selected by ledger id.
+ */
+ private void executeOrdered(Executor callbackExecutor, long ledgerId,
Runnable r) {
+ PerChannelBookieClient.executeOrdered(executor, callbackExecutor,
ledgerId, r);
+ }
+
private int getRc(int rc) {
if (BKException.Code.OK == rc) {
return rc;
@@ -224,7 +233,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
@Override
public void forceLedger(final BookieId addr, final long ledgerId,
- final ForceLedgerCallback cb, final Object ctx) {
+ final ForceLedgerCallback cb, final Object ctx, final Executor
callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
cb.forceLedgerComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -235,20 +244,21 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
client.obtain((rc, pcbc) -> {
if (rc != BKException.Code.OK) {
try {
- executor.executeOrdered(ledgerId,
+ executeOrdered(callbackExecutor, ledgerId,
() -> cb.forceLedgerComplete(rc, ledgerId, addr,
ctx));
} catch (RejectedExecutionException re) {
cb.forceLedgerComplete(getRc(BKException.Code.InterruptedException), ledgerId,
addr, ctx);
}
} else {
- pcbc.forceLedger(ledgerId, cb, ctx);
+ pcbc.forceLedger(ledgerId, cb, ctx, callbackExecutor);
}
}, ledgerId);
}
@Override
public void writeLac(final BookieId addr, final long ledgerId, final
byte[] masterKey,
- final long lac, final ByteBufList toSend, final WriteLacCallback
cb, final Object ctx) {
+ final long lac, final ByteBufList toSend, final WriteLacCallback
cb, final Object ctx,
+ final Executor callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
cb.writeLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -261,13 +271,13 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
try {
if (rc != BKException.Code.OK) {
try {
- executor.executeOrdered(ledgerId,
+ executeOrdered(callbackExecutor, ledgerId,
() -> cb.writeLacComplete(rc, ledgerId, addr,
ctx));
} catch (RejectedExecutionException re) {
cb.writeLacComplete(getRc(BKException.Code.InterruptedException), ledgerId,
addr, ctx);
}
} else {
- pcbc.writeLac(ledgerId, masterKey, lac, toSend, cb, ctx);
+ pcbc.writeLac(ledgerId, masterKey, lac, toSend, cb, ctx,
callbackExecutor);
}
} finally {
ReferenceCountUtil.release(toSend);
@@ -294,7 +304,8 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
final Object ctx,
final int options,
final boolean allowFastFail,
- final EnumSet<WriteFlag> writeFlags) {
+ final EnumSet<WriteFlag> writeFlags,
+ final Executor callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
completeAdd(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -308,7 +319,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
client.obtain(ChannelReadyForAddEntryCallback.create(
this, toSend, ledgerId, entryId, addr,
- ctx, cb, options, masterKey, allowFastFail,
writeFlags),
+ ctx, cb, options, masterKey, allowFastFail,
writeFlags, callbackExecutor),
ledgerId);
}
@@ -344,9 +355,11 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
final long entryId,
final ByteBuf entry,
final ReadEntryCallback cb,
- final Object ctx) {
+ final Object ctx,
+ final Executor callbackExecutor) {
try {
- executor.executeOrdered(ledgerId, () -> cb.readEntryComplete(rc,
ledgerId, entryId, entry, ctx));
+ executeOrdered(callbackExecutor, ledgerId,
+ () -> cb.readEntryComplete(rc, ledgerId, entryId, entry,
ctx));
} catch (RejectedExecutionException ree) {
cb.readEntryComplete(getRc(BKException.Code.InterruptedException),
ledgerId, entryId, entry, ctx);
@@ -358,9 +371,11 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
final long startEntryId,
final ByteBufList bufList,
final BatchedReadEntryCallback cb,
- final Object ctx) {
+ final Object ctx,
+ final Executor callbackExecutor) {
try {
- executor.executeOrdered(ledgerId, () -> cb.readEntriesComplete(rc,
ledgerId, startEntryId, bufList, ctx));
+ executeOrdered(callbackExecutor, ledgerId,
+ () -> cb.readEntriesComplete(rc, ledgerId, startEntryId,
bufList, ctx));
} catch (RejectedExecutionException ree) {
cb.readEntriesComplete(getRc(BKException.Code.InterruptedException),
ledgerId, startEntryId, bufList, ctx);
@@ -386,12 +401,13 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
private byte[] masterKey;
private boolean allowFastFail;
private EnumSet<WriteFlag> writeFlags;
+ private Executor callbackExecutor;
static ChannelReadyForAddEntryCallback create(
BookieClientImpl bookieClient, ReferenceCounted toSend, long
ledgerId,
long entryId, BookieId addr, Object ctx,
WriteCallback cb, int options, byte[] masterKey, boolean
allowFastFail,
- EnumSet<WriteFlag> writeFlags) {
+ EnumSet<WriteFlag> writeFlags, Executor callbackExecutor) {
ChannelReadyForAddEntryCallback callback = RECYCLER.get();
callback.bookieClient = bookieClient;
callback.toSend = toSend;
@@ -404,6 +420,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
callback.masterKey = masterKey;
callback.allowFastFail = allowFastFail;
callback.writeFlags = writeFlags;
+ callback.callbackExecutor = callbackExecutor;
return callback;
}
@@ -411,7 +428,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
public void operationComplete(final int rc,
PerChannelBookieClient pcbc) {
if (rc != BKException.Code.OK) {
- bookieClient.executor.executeOrdered(ledgerId, () -> {
+ bookieClient.executeOrdered(callbackExecutor, ledgerId, () -> {
try {
bookieClient.completeAdd(rc, ledgerId, entryId, addr,
cb, ctx);
} finally {
@@ -422,7 +439,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
} else {
try {
pcbc.addEntry(ledgerId, masterKey, entryId,
- toSend, cb, ctx, options, allowFastFail,
writeFlags);
+ toSend, cb, ctx, options, allowFastFail,
writeFlags, callbackExecutor);
} finally {
ReferenceCountUtil.release(toSend);
}
@@ -456,13 +473,14 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
masterKey = null;
allowFastFail = false;
writeFlags = null;
+ callbackExecutor = null;
recyclerHandle.recycle(this);
}
}
@Override
public void readLac(final BookieId addr, final long ledgerId, final
ReadLacCallback cb,
- final Object ctx) {
+ final Object ctx, final Executor callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
cb.readLacComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
ledgerId, null, null,
@@ -472,14 +490,14 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
client.obtain((rc, pcbc) -> {
if (rc != BKException.Code.OK) {
try {
- executor.executeOrdered(ledgerId,
+ executeOrdered(callbackExecutor, ledgerId,
() -> cb.readLacComplete(rc, ledgerId, null, null,
ctx));
} catch (RejectedExecutionException re) {
cb.readLacComplete(getRc(BKException.Code.InterruptedException),
ledgerId, null, null, ctx);
}
} else {
- pcbc.readLac(ledgerId, cb, ctx);
+ pcbc.readLac(ledgerId, cb, ctx, callbackExecutor);
}
}, ledgerId, useV3Enforced);
}
@@ -499,7 +517,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
@Override
public void readEntry(final BookieId addr, final long ledgerId, final long
entryId,
final ReadEntryCallback cb, final Object ctx, int
flags, byte[] masterKey,
- final boolean allowFastFail) {
+ final boolean allowFastFail, final Executor
callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
cb.readEntryComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -509,9 +527,9 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
client.obtain((rc, pcbc) -> {
if (rc != BKException.Code.OK) {
- completeRead(rc, ledgerId, entryId, null, cb, ctx);
+ completeRead(rc, ledgerId, entryId, null, cb, ctx,
callbackExecutor);
} else {
- pcbc.readEntry(ledgerId, entryId, cb, ctx, flags, masterKey,
allowFastFail);
+ pcbc.readEntry(ledgerId, entryId, cb, ctx, flags, masterKey,
allowFastFail, callbackExecutor);
}
}, ledgerId);
}
@@ -519,7 +537,7 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
@Override
public void batchReadEntries(final BookieId address, final long ledgerId,
final long startEntryId,
final int maxCount, final long maxSize, final
BatchedReadEntryCallback cb, final Object ctx,
- final int flags, final byte[] masterKey, final boolean
allowFastFail) {
+ final int flags, final byte[] masterKey, final boolean
allowFastFail, final Executor callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(address);
if (client == null) {
cb.readEntriesComplete(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -529,10 +547,10 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
client.obtain((rc, pcbc) -> {
if (rc != BKException.Code.OK) {
- completeBatchRead(rc, ledgerId, startEntryId, null, cb, ctx);
+ completeBatchRead(rc, ledgerId, startEntryId, null, cb, ctx,
callbackExecutor);
} else {
pcbc.batchReadEntries(ledgerId, startEntryId, maxCount,
maxSize, cb, ctx, flags, masterKey,
- allowFastFail);
+ allowFastFail, callbackExecutor);
}
}, ledgerId);
}
@@ -545,20 +563,21 @@ public class BookieClientImpl implements BookieClient,
PerChannelBookieClientFac
final long timeOutInMillis,
final boolean piggyBackEntry,
final ReadEntryCallback cb,
- final Object ctx) {
+ final Object ctx,
+ final Executor callbackExecutor) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
completeRead(BKException.Code.BookieHandleNotAvailableException,
- ledgerId, entryId, null, cb, ctx);
+ ledgerId, entryId, null, cb, ctx, callbackExecutor);
return;
}
client.obtain((rc, pcbc) -> {
if (rc != BKException.Code.OK) {
- completeRead(rc, ledgerId, entryId, null, cb, ctx);
+ completeRead(rc, ledgerId, entryId, null, cb, ctx,
callbackExecutor);
} else {
pcbc.readEntryWaitForLACUpdate(ledgerId, entryId, previousLAC,
timeOutInMillis, piggyBackEntry, cb,
- ctx);
+ ctx, callbackExecutor);
}
}, ledgerId);
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/CompletionValue.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/CompletionValue.java
index c01103fdcf..233df9264f 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/CompletionValue.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/CompletionValue.java
@@ -24,6 +24,7 @@ package org.apache.bookkeeper.proto;
import io.github.merlimat.slog.Event;
import io.netty.channel.Channel;
import java.util.Map;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import lombok.CustomLog;
import org.apache.bookkeeper.client.BKException;
@@ -43,16 +44,26 @@ abstract class CompletionValue {
protected OpStatsLogger timeoutOpLogger;
protected Map<String, String> mdcContextMap;
protected PerChannelBookieClient perChannelBookieClient;
+ /** Executor running the callback; {@code null} means the worker thread
selected by ledger id. */
+ protected Executor callbackExecutor;
public CompletionValue(String operationName,
Object ctx,
long ledgerId, long entryId, PerChannelBookieClient
perChannelBookieClient) {
+ this(operationName, ctx, ledgerId, entryId, perChannelBookieClient,
null);
+ }
+
+ public CompletionValue(String operationName,
+ Object ctx,
+ long ledgerId, long entryId, PerChannelBookieClient
perChannelBookieClient,
+ Executor callbackExecutor) {
this.operationName = operationName;
this.ctx = ctx;
this.ledgerId = ledgerId;
this.entryId = entryId;
this.startTime = MathUtils.nowInNano();
this.perChannelBookieClient = perChannelBookieClient;
+ this.callbackExecutor = callbackExecutor;
if (perChannelBookieClient != null) {
this.mdcContextMap =
perChannelBookieClient.preserveMdcForTaskExecution ? MDC.getCopyOfContextMap()
: null;
}
@@ -154,8 +165,16 @@ abstract class CompletionValue {
// no-op
}
+ /**
+ * Runs {@code r} on the thread owning the callbacks of this operation:
the caller-supplied executor when
+ * present, otherwise the worker thread selected by ledger id.
+ */
+ void executeOrdered(Runnable r) {
+ PerChannelBookieClient.executeOrdered(perChannelBookieClient.executor,
callbackExecutor, ledgerId, r);
+ }
+
protected void errorOutAndRunCallback(final Runnable callback) {
- perChannelBookieClient.executor.executeOrdered(ledgerId, () -> {
+ executeOrdered(() -> {
String bAddress = "null";
Channel c = perChannelBookieClient.channel;
if (c != null && c.remoteAddress() != null) {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ForceLedgerCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ForceLedgerCompletion.java
index 34bd0384fb..7285c55e01 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ForceLedgerCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ForceLedgerCompletion.java
@@ -21,6 +21,7 @@
package org.apache.bookkeeper.proto;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.BKException;
class ForceLedgerCompletion extends CompletionValue {
@@ -30,9 +31,10 @@ class ForceLedgerCompletion extends CompletionValue {
final
BookkeeperInternalCallbacks.ForceLedgerCallback originalCallback,
final Object originalCtx,
final long ledgerId,
- PerChannelBookieClient
perChannelBookieClient) {
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
super("ForceLedger",
- originalCtx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient);
+ originalCtx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient, callbackExecutor);
this.opLogger = perChannelBookieClient.forceLedgerOpLogger;
this.timeoutOpLogger =
perChannelBookieClient.forceLedgerTimeoutOpLogger;
this.cb = (rc, ledgerId1, addr, ctx) -> {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
index e6380bff59..7ec0900904 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
@@ -76,6 +76,7 @@ import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
@@ -681,14 +682,14 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
}
void writeLac(final long ledgerId, final byte[] masterKey, final long lac,
ByteBufList toSend, WriteLacCallback cb,
- Object ctx) {
+ Object ctx, Executor callbackExecutor) {
final long txnId = getTxnId();
final CompletionKey completionKey = new TxnCompletionKey(txnId,
OperationType.WRITE_LAC);
// writeLac is mostly like addEntry hence uses addEntryTimeout
completionObjects.put(completionKey,
new WriteLacCompletion(completionKey, cb,
- ctx, ledgerId, this));
+ ctx, ledgerId, this,
callbackExecutor));
// Build the request
Request writeLacRequest = new Request();
@@ -708,10 +709,10 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
writeAndFlush(channel, completionKey, writeLacRequest, false,
releaseBody, releaseBody);
}
- void forceLedger(final long ledgerId, ForceLedgerCallback cb, Object ctx) {
+ void forceLedger(final long ledgerId, ForceLedgerCallback cb, Object ctx,
Executor callbackExecutor) {
if (useV2WireProtocol) {
log.error("force is not allowed with v2 protocol");
- executor.executeOrdered(ledgerId, () -> {
+ executeOrdered(executor, callbackExecutor, ledgerId, () -> {
cb.forceLedgerComplete(BKException.Code.IllegalOpException, ledgerId, bookieId,
ctx);
});
return;
@@ -722,7 +723,7 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
// force is mostly like addEntry hence uses addEntryTimeout
completionObjects.put(completionKey,
new ForceLedgerCompletion(completionKey, cb,
- ctx, ledgerId, this));
+ ctx, ledgerId, this,
callbackExecutor));
// Build the request
Request forceLedgerRequest = new Request();
@@ -755,9 +756,12 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
* allowFastFail flag
* @param writeFlags
* WriteFlags
+ * @param callbackExecutor
+ * Executor running the callback, or null for the worker thread
selected by ledgerId
*/
void addEntry(final long ledgerId, byte[] masterKey, final long entryId,
ReferenceCounted toSend, WriteCallback cb,
- Object ctx, final int options, boolean allowFastFail, final
EnumSet<WriteFlag> writeFlags) {
+ Object ctx, final int options, boolean allowFastFail, final
EnumSet<WriteFlag> writeFlags,
+ Executor callbackExecutor) {
Object request = null;
CompletionKey completionKey = null;
Runnable cleanupActionFailedBeforeWrite = null;
@@ -819,13 +823,13 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
putCompletionKeyValue(completionKey,
AddCompletion.acquireAddCompletion(completionKey,
- cb, ctx, ledgerId, entryId,
this));
+ cb, ctx, ledgerId, entryId,
this, callbackExecutor));
// addEntry times out on backpressure
writeAndFlush(channel, completionKey, request, allowFastFail,
cleanupActionFailedBeforeWrite,
cleanupActionAfterWrite);
}
- public void readLac(final long ledgerId, ReadLacCallback cb, Object ctx) {
+ public void readLac(final long ledgerId, ReadLacCallback cb, Object ctx,
Executor callbackExecutor) {
Object request = null;
CompletionKey completionKey = null;
if (useV2WireProtocol) {
@@ -848,7 +852,7 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
}
putCompletionKeyValue(completionKey,
new ReadLacCompletion(completionKey, cb,
- ctx, ledgerId, this));
+ ctx, ledgerId, this,
callbackExecutor));
writeAndFlush(channel, completionKey, request);
}
@@ -878,9 +882,10 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
final long timeOutInMillis,
final boolean piggyBackEntry,
ReadEntryCallback cb,
- Object ctx) {
+ Object ctx,
+ Executor callbackExecutor) {
readEntryInternal(ledgerId, entryId, previousLAC, timeOutInMillis,
- piggyBackEntry, cb, ctx, (short) 0, null, false);
+ piggyBackEntry, cb, ctx, (short) 0, null, false,
callbackExecutor);
}
/**
@@ -892,9 +897,10 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
Object ctx,
int flags,
byte[] masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail,
+ Executor callbackExecutor) {
readEntryInternal(ledgerId, entryId, null, null, false,
- cb, ctx, (short) flags, masterKey, allowFastFail);
+ cb, ctx, (short) flags, masterKey, allowFastFail,
callbackExecutor);
}
private void readEntryInternal(final long ledgerId,
@@ -906,7 +912,8 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
final Object ctx,
int flags,
byte[] masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail,
+ Executor callbackExecutor) {
Object request = null;
CompletionKey completionKey = null;
if (useV2WireProtocol) {
@@ -970,7 +977,8 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
request = readEntryRequest;
}
- ReadCompletion readCompletion = new ReadCompletion(completionKey, cb,
ctx, ledgerId, entryId, this);
+ ReadCompletion readCompletion = new ReadCompletion(completionKey, cb,
ctx, ledgerId, entryId, this,
+ callbackExecutor);
putCompletionKeyValue(completionKey, readCompletion);
writeAndFlush(channel, completionKey, request, allowFastFail, null,
null);
@@ -984,10 +992,11 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
Object ctx,
int flags,
byte[] masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail,
+ Executor callbackExecutor) {
batchReadEntriesInternal(ledgerId, startEntryId, maxCount, maxSize,
null, null, false,
- cb, ctx, (short) flags, masterKey, allowFastFail);
+ cb, ctx, (short) flags, masterKey, allowFastFail,
callbackExecutor);
}
private void batchReadEntriesInternal(final long ledgerId,
@@ -1001,7 +1010,8 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
final Object ctx,
int flags,
byte[] masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail,
+ Executor callbackExecutor) {
Object request;
CompletionKey completionKey;
final long txnId = getTxnId();
@@ -1013,7 +1023,7 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
throw new UnsupportedOperationException("Unsupported batch read
entry operation for v3 protocol.");
}
BatchedReadCompletion readCompletion = new BatchedReadCompletion(
- completionKey, cb, ctx, ledgerId, startEntryId, this);
+ completionKey, cb, ctx, ledgerId, startEntryId, this,
callbackExecutor);
putCompletionKeyValue(completionKey, readCompletion);
writeAndFlush(channel, completionKey, request, allowFastFail, null,
null);
@@ -1382,6 +1392,18 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
}
}
+ /**
+ * Runs {@code r} on the thread owning the callbacks of a ledger: the
caller-supplied
+ * {@code callbackExecutor} when present, otherwise the worker thread
{@code executor} selects by ledger id.
+ */
+ static void executeOrdered(OrderedExecutor executor, Executor
callbackExecutor, long ledgerId, Runnable r) {
+ if (callbackExecutor != null) {
+ callbackExecutor.execute(r);
+ } else {
+ executor.executeOrdered(ledgerId, r);
+ }
+ }
+
private void readV2Response(final BookieProtocol.Response response) {
OperationType operationType = getOperationType(response.getOpCode());
StatusCode status = getStatusCodeFromErrorCode(response.errorCode);
@@ -1402,8 +1424,7 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
.log("Unexpected response received from bookie");
response.release();
} else {
- long orderingKey = completionValue.ledgerId;
- executor.executeOrdered(orderingKey,
+ completionValue.executeOrdered(
ReadV2ResponseCallback.create(completionValue,
response.ledgerId, response.entryId,
status, response));
}
@@ -1518,8 +1539,7 @@ public class PerChannelBookieClient extends
ChannelInboundHandlerAdapter {
.attr("txnId", () -> header.getTxnId())
.log("Unexpected response received from bookie");
} else {
- long orderingKey = completionValue.ledgerId;
- executor.executeOrdered(orderingKey, new Runnable() {
+ completionValue.executeOrdered(new Runnable() {
@Override
public void run() {
completionValue.restoreMdcContext();
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadCompletion.java
index 828bfccf57..8e7d695394 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadCompletion.java
@@ -26,6 +26,7 @@ import static
org.apache.bookkeeper.client.LedgerHandle.INVALID_ENTRY_ID;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.ReferenceCountUtil;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.BKException;
class ReadCompletion extends CompletionValue {
@@ -35,8 +36,9 @@ class ReadCompletion extends CompletionValue {
final BookkeeperInternalCallbacks.ReadEntryCallback
originalCallback,
final Object originalCtx,
long ledgerId, final long entryId,
- PerChannelBookieClient perChannelBookieClient) {
- super("Read", originalCtx, ledgerId, entryId, perChannelBookieClient);
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
+ super("Read", originalCtx, ledgerId, entryId, perChannelBookieClient,
callbackExecutor);
this.opLogger = perChannelBookieClient.readEntryOpLogger;
this.timeoutOpLogger = perChannelBookieClient.readTimeoutOpLogger;
this.cb = (rc, ledgerId1, entryId1, buffer, ctx) -> {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacCompletion.java
index 73c734c6f2..73efac34bf 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacCompletion.java
@@ -23,6 +23,7 @@ package org.apache.bookkeeper.proto;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.BKException;
class ReadLacCompletion extends CompletionValue {
@@ -31,8 +32,9 @@ class ReadLacCompletion extends CompletionValue {
public ReadLacCompletion(final CompletionKey key,
BookkeeperInternalCallbacks.ReadLacCallback
originalCallback,
final Object ctx, final long ledgerId,
- PerChannelBookieClient perChannelBookieClient) {
- super("ReadLAC", ctx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient);
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
+ super("ReadLAC", ctx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient, callbackExecutor);
this.opLogger = perChannelBookieClient.readLacOpLogger;
this.timeoutOpLogger = perChannelBookieClient.readLacTimeoutOpLogger;
this.cb = new BookkeeperInternalCallbacks.ReadLacCallback() {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacCompletion.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacCompletion.java
index ffbefb0f9d..dc0ed6146d 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacCompletion.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacCompletion.java
@@ -21,6 +21,7 @@
package org.apache.bookkeeper.proto;
+import java.util.concurrent.Executor;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.net.BookieId;
@@ -31,9 +32,10 @@ class WriteLacCompletion extends CompletionValue {
final
BookkeeperInternalCallbacks.WriteLacCallback originalCallback,
final Object originalCtx,
final long ledgerId,
- PerChannelBookieClient perChannelBookieClient) {
+ PerChannelBookieClient perChannelBookieClient,
+ Executor callbackExecutor) {
super("WriteLAC",
- originalCtx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient);
+ originalCtx, ledgerId, BookieProtocol.LAST_ADD_CONFIRMED,
perChannelBookieClient, callbackExecutor);
this.opLogger = perChannelBookieClient.writeLacOpLogger;
this.timeoutOpLogger = perChannelBookieClient.writeLacTimeoutOpLogger;
this.cb = new BookkeeperInternalCallbacks.WriteLacCallback() {
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
index 3090cd5134..ffa59835ba 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
@@ -546,6 +546,10 @@ public abstract class MockBookKeeperTestCase {
stub.when(bookieClient).readEntry(any(), anyLong(), anyLong(),
any(BookkeeperInternalCallbacks.ReadEntryCallback.class),
any(), anyInt(), any(), anyBoolean());
+
+ stub.when(bookieClient).readEntry(any(), anyLong(), anyLong(),
+ any(BookkeeperInternalCallbacks.ReadEntryCallback.class),
+ any(), anyInt(), any(), anyBoolean(), any());
}
protected void setupBookieClientBatchReadEntry() {
@@ -624,6 +628,8 @@ public abstract class MockBookKeeperTestCase {
any(), any(), anyInt(), any());
stub.when(bookieClient).batchReadEntries(any(BookieId.class),
anyLong(), anyLong(), anyInt(), anyLong(),
any(), any(), anyInt(), any(), anyBoolean());
+ stub.when(bookieClient).batchReadEntries(any(BookieId.class),
anyLong(), anyLong(), anyInt(), anyLong(),
+ any(), any(), anyInt(), any(), anyBoolean(), any());
}
@SuppressWarnings("unchecked")
@@ -651,6 +657,9 @@ public abstract class MockBookKeeperTestCase {
stub.when(bookieClient).readLac(any(BookieId.class), anyLong(),
any(BookkeeperInternalCallbacks.ReadLacCallback.class),
any());
+ stub.when(bookieClient).readLac(any(BookieId.class), anyLong(),
+ any(BookkeeperInternalCallbacks.ReadLacCallback.class),
+ any(), any());
}
private byte[] extractEntryPayload(long ledgerId, long entryId,
ByteBufList toSend)
@@ -726,6 +735,11 @@ public abstract class MockBookKeeperTestCase {
anyLong(), any(ByteBufList.class),
any(BookkeeperInternalCallbacks.WriteCallback.class),
any(), anyInt(), anyBoolean(), any(EnumSet.class));
+ stub.when(bookieClient).addEntry(any(BookieId.class),
+ anyLong(), any(byte[].class),
+ anyLong(), any(ByteBufList.class),
+ any(BookkeeperInternalCallbacks.WriteCallback.class),
+ any(), anyInt(), anyBoolean(), any(EnumSet.class), any());
}
@SuppressWarnings("unchecked")
@@ -767,6 +781,10 @@ public abstract class MockBookKeeperTestCase {
anyLong(),
any(BookkeeperInternalCallbacks.ForceLedgerCallback.class),
any());
+ stub.when(bookieClient).forceLedger(any(BookieId.class),
+ anyLong(),
+ any(BookkeeperInternalCallbacks.ForceLedgerCallback.class),
+ any(), any());
}
}
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingWriteLacOpTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingWriteLacOpTest.java
index 0a8fa83b70..e04ab46952 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingWriteLacOpTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingWriteLacOpTest.java
@@ -52,7 +52,8 @@ public class PendingWriteLacOpTest implements
AsyncCallback.AddLacCallback {
mockClientContext = mock(ClientContext.class);
mockBookieClient = mock(BookieClient.class);
doNothing().when(mockBookieClient).writeLac(any(BookieId.class),
anyLong(), any(byte[].class), anyLong(),
- any(ByteBufList.class),
any(BookkeeperInternalCallbacks.WriteLacCallback.class), any(Object.class));
+ any(ByteBufList.class),
any(BookkeeperInternalCallbacks.WriteLacCallback.class), any(Object.class),
+ any());
when(mockClientContext.getBookieClient()).thenReturn(mockBookieClient);
callbackInvoked = false;
}
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOpTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOpTest.java
index 760f249018..37fb1d88ce 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOpTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedAndEntryOpTest.java
@@ -138,6 +138,9 @@ public class ReadLastConfirmedAndEntryOpTest {
when(mockLh.getDistributionSchedule()).thenReturn(distributionSchedule);
digestManager = new DummyDigestManager(LEDGERID, false,
UnpooledByteBufAllocator.DEFAULT);
when(mockLh.getDigestManager()).thenReturn(digestManager);
+ // the op issues its speculative reads through the handle's executor
+ when(mockLh.submitOrdered(any())).thenAnswer(
+ invocation -> orderedScheduler.submitOrdered(LEDGERID,
invocation.getArgument(0)));
}
@After
@@ -198,6 +201,7 @@ public class ReadLastConfirmedAndEntryOpTest {
anyLong(),
anyBoolean(),
any(ReadEntryCallback.class),
+ any(),
any()
);
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersOpenLedgerTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersOpenLedgerTest.java
index 177da38e92..550a32e0c8 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersOpenLedgerTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersOpenLedgerTest.java
@@ -23,6 +23,7 @@ package org.apache.bookkeeper.client.api;
import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.doAnswer;
@@ -130,7 +131,7 @@ public class BookKeeperBuildersOpenLedgerTest extends
MockBookKeeperTestCase {
return null;
}).when(bookieClient).readEntry(any(BookieId.class),
anyLong(), anyLong(),
any(BookkeeperInternalCallbacks.ReadEntryCallback.class),
- any(), anyInt(), any());
+ any(), anyInt(), any(), anyBoolean(), any());
// Mock read lac.
doAnswer(invocation -> {
long ledgerId = (long) invocation.getArguments()[1];
@@ -141,7 +142,7 @@ public class BookKeeperBuildersOpenLedgerTest extends
MockBookKeeperTestCase {
return null;
}).when(bookieClient).readLac(any(BookieId.class),
anyLong(),
any(BookkeeperInternalCallbacks.ReadLacCallback.class),
- any());
+ any(), any());
}
private void resetBKClient() throws Exception {
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/OrderingKeyTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/OrderingKeyTest.java
new file mode 100644
index 0000000000..543cb9fd06
--- /dev/null
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/OrderingKeyTest.java
@@ -0,0 +1,209 @@
+/*
+ *
+ * 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.client.api;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerHandleAdv;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that {@link CreateBuilder#withOrderingKey(Object)} and {@link
OpenBuilder#withOrderingKey(Object)} pin
+ * every callback of the resulting handle to the worker thread selected by the
key, and that without a key the
+ * thread is still selected by ledger id. Both wire protocols are covered
since their response dispatch differs.
+ */
+public class OrderingKeyTest extends BookKeeperClusterTestCase {
+
+ private static final byte[] PASSWORD = "ordering-key".getBytes(UTF_8);
+ private static final byte[] DATA = "entry".getBytes(UTF_8);
+ private static final int NUM_WORKER_THREADS = 4;
+ private static final long TIMEOUT_SECONDS = 30;
+
+ public OrderingKeyTest() {
+ super(3);
+ }
+
+ @Test
+ public void testOrderingKeyV3() throws Exception {
+ testOrderingKey(false);
+ }
+
+ @Test
+ public void testOrderingKeyV2() throws Exception {
+ testOrderingKey(true);
+ }
+
+ @Test
+ public void testDefaultKeyedByLedgerIdV3() throws Exception {
+ testDefaultKeyedByLedgerId(false);
+ }
+
+ @Test
+ public void testDefaultKeyedByLedgerIdV2() throws Exception {
+ testDefaultKeyedByLedgerId(true);
+ }
+
+ private void testOrderingKey(boolean useV2WireProtocol) throws Exception {
+ try (BookKeeper bk = new BookKeeper(clientConf(useV2WireProtocol))) {
+ OrderedExecutor pool = bk.getMainWorkerPool();
+
+ // Create with an explicit ledger id, so that the key can be
chosen to map to a different
+ // thread than the ledger id: this proves the key, not the id,
selects the thread.
+ long ledgerId = 0xABCDEFL;
+ String createKey = keyOnOtherThread(pool, "create", ledgerId);
+ Thread createKeyThread = threadOf(pool.chooseThread(createKey));
+ assertNotSame(threadOf(pool.chooseThread(ledgerId)),
createKeyThread);
+
+ LedgerHandleAdv writer = (LedgerHandleAdv) bk.newCreateLedgerOp()
+
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
+ .withPassword(PASSWORD)
+ .makeAdv()
+ .withLedgerId(ledgerId)
+ .withOrderingKey(createKey)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertEquals(ledgerId, writer.getId());
+ for (long entryId = 0; entryId < 3; entryId++) {
+ long id = entryId;
+ assertSame(createKeyThread, callbackThread(
+ f -> writer.asyncAddEntry(id, DATA, (rc, lh, eid, ctx)
-> complete(f, rc), null)));
+ }
+ assertSame(createKeyThread, callbackThread(
+ f -> writer.asyncReadEntries(0, 2, (rc, lh, entries, ctx)
-> complete(f, rc), null)));
+ writer.close();
+
+ // Open the closed ledger, without recovery, under a different key.
+ String openKey = keyOnOtherThread(pool, "open", ledgerId);
+ Thread openKeyThread = threadOf(pool.chooseThread(openKey));
+ LedgerHandle reader = (LedgerHandle) bk.newOpenLedgerOp()
+
.withLedgerId(ledgerId).withPassword(PASSWORD).withRecovery(false)
+ .withOrderingKey(openKey)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertSame(openKeyThread, callbackThread(
+ f -> reader.asyncReadEntries(0, 2, (rc, lh, entries, ctx)
-> complete(f, rc), null)));
+ reader.close();
+
+ // Open an unclosed ledger without recovery: reading the last add
confirmed goes to the bookies
+ // (on a closed ledger it completes inline from the metadata) and
lands on the key's thread.
+ LedgerHandle unclosed = (LedgerHandle) bk.newCreateLedgerOp()
+
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
+ .withPassword(PASSWORD)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ for (int i = 0; i < 3; i++) {
+ unclosed.addEntry(DATA);
+ }
+ String lacKey = keyOnOtherThread(pool, "lac", unclosed.getId());
+ Thread lacKeyThread = threadOf(pool.chooseThread(lacKey));
+ LedgerHandle tailer = (LedgerHandle) bk.newOpenLedgerOp()
+
.withLedgerId(unclosed.getId()).withPassword(PASSWORD).withRecovery(false)
+ .withOrderingKey(lacKey)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertSame(lacKeyThread, callbackThread(
+ f -> tailer.asyncReadLastConfirmed((rc, lac, ctx) ->
complete(f, rc), null)));
+ tailer.close();
+
+ // Open the same unclosed ledger with recovery: the recovery reads
and adds, the recovery
+ // completion and the reads that follow all run under the key. The
fenced writer is
+ // deliberately left open.
+ String recoveryKey = keyOnOtherThread(pool, "recovery",
unclosed.getId());
+ Thread recoveryKeyThread =
threadOf(pool.chooseThread(recoveryKey));
+ LedgerHandle recovered = (LedgerHandle) bk.newOpenLedgerOp()
+
.withLedgerId(unclosed.getId()).withPassword(PASSWORD).withRecovery(true)
+ .withOrderingKey(recoveryKey)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertEquals(2, recovered.getLastAddConfirmed());
+ assertSame(recoveryKeyThread, callbackThread(
+ f -> recovered.asyncReadEntries(0, 2, (rc, lh, entries,
ctx) -> complete(f, rc), null)));
+ recovered.close();
+ }
+ }
+
+ private void testDefaultKeyedByLedgerId(boolean useV2WireProtocol) throws
Exception {
+ try (BookKeeper bk = new BookKeeper(clientConf(useV2WireProtocol))) {
+ OrderedExecutor pool = bk.getMainWorkerPool();
+
+ LedgerHandle writer = (LedgerHandle) bk.newCreateLedgerOp()
+
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
+ .withPassword(PASSWORD)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ Thread ledgerIdThread =
threadOf(pool.chooseThread(writer.getId()));
+ assertSame(ledgerIdThread, callbackThread(
+ f -> writer.asyncAddEntry(DATA, (rc, lh, eid, ctx) ->
complete(f, rc), null)));
+ writer.close();
+
+ LedgerHandle reader = (LedgerHandle) bk.newOpenLedgerOp()
+ .withLedgerId(writer.getId()).withPassword(PASSWORD)
+ .execute().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertSame(ledgerIdThread, callbackThread(
+ f -> reader.asyncReadEntries(0, 0, (rc, lh, entries, ctx)
-> complete(f, rc), null)));
+ reader.close();
+ }
+ }
+
+ private ClientConfiguration clientConf(boolean useV2WireProtocol) {
+ return new ClientConfiguration(baseClientConf)
+ .setUseV2WireProtocol(useV2WireProtocol)
+ .setNumWorkerThreads(NUM_WORKER_THREADS);
+ }
+
+ /** Picks a key whose worker thread differs from the one the pool selects
for {@code ledgerId}. */
+ private static String keyOnOtherThread(OrderedExecutor pool, String
prefix, long ledgerId) {
+ for (int i = 0; i < 100; i++) {
+ String key = prefix + "-" + i;
+ if (pool.chooseThread(key) != pool.chooseThread(ledgerId)) {
+ return key;
+ }
+ }
+ throw new AssertionError("no key mapped to a thread other than the
ledger id's");
+ }
+
+ /** The thread behind one of the pool's single-threaded executors. */
+ private static Thread threadOf(Executor executor) throws Exception {
+ return callbackThread(f -> executor.execute(() ->
f.complete(Thread.currentThread())));
+ }
+
+ /** Runs {@code operation} and returns the thread on which it completed
the future. */
+ private static Thread callbackThread(Consumer<CompletableFuture<Thread>>
operation) throws Exception {
+ CompletableFuture<Thread> thread = new CompletableFuture<>();
+ operation.accept(thread);
+ return thread.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+
+ private static void complete(CompletableFuture<Thread> future, int rc) {
+ if (rc == BKException.Code.OK) {
+ future.complete(Thread.currentThread());
+ } else {
+ future.completeExceptionally(BKException.create(rc));
+ }
+ }
+}
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookieClient.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookieClient.java
index 3731731747..d2da2d2c62 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookieClient.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookieClient.java
@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
import lombok.Getter;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.api.WriteFlag;
@@ -130,6 +131,11 @@ public class MockBookieClient implements BookieClient {
return mockBookies;
}
+ /** Executor running a callback: the caller-supplied one, or the worker
thread selected by ledger id. */
+ private Executor executorFor(Executor callbackExecutor, long ledgerId) {
+ return callbackExecutor != null ? callbackExecutor :
executor.chooseThread(ledgerId);
+ }
+
@Override
public List<BookieId> getFaultyBookies() {
return Collections.emptyList();
@@ -147,22 +153,23 @@ public class MockBookieClient implements BookieClient {
@Override
public void forceLedger(BookieId addr, long ledgerId,
- ForceLedgerCallback cb, Object ctx) {
- executor.executeOrdered(ledgerId,
+ ForceLedgerCallback cb, Object ctx, Executor
callbackExecutor) {
+ executorFor(callbackExecutor, ledgerId).execute(
() ->
cb.forceLedgerComplete(BKException.Code.IllegalOpException, ledgerId, addr,
ctx));
}
@Override
public void writeLac(BookieId addr, long ledgerId, byte[] masterKey,
- long lac, ByteBufList toSend, WriteLacCallback cb,
Object ctx) {
- executor.executeOrdered(ledgerId,
+ long lac, ByteBufList toSend, WriteLacCallback cb,
Object ctx, Executor callbackExecutor) {
+ executorFor(callbackExecutor, ledgerId).execute(
() -> cb.writeLacComplete(BKException.Code.IllegalOpException,
ledgerId, addr, ctx));
}
@Override
public void addEntry(BookieId addr, long ledgerId, byte[] masterKey,
long entryId, ReferenceCounted toSend, WriteCallback
cb, Object ctx,
- int options, boolean allowFastFail,
EnumSet<WriteFlag> writeFlags) {
+ int options, boolean allowFastFail,
EnumSet<WriteFlag> writeFlags,
+ Executor callbackExecutor) {
toSend.retain();
preWriteHook.runHook(addr, ledgerId, entryId)
.thenComposeAsync(
@@ -195,19 +202,19 @@ public class MockBookieClient implements BookieClient {
} else {
cb.writeComplete(BKException.Code.OK, ledgerId,
entryId, addr, ctx);
}
- }, executor.chooseThread(ledgerId));
+ }, executorFor(callbackExecutor, ledgerId));
}
@Override
- public void readLac(BookieId addr, long ledgerId, ReadLacCallback cb,
Object ctx) {
- executor.executeOrdered(ledgerId,
+ public void readLac(BookieId addr, long ledgerId, ReadLacCallback cb,
Object ctx, Executor callbackExecutor) {
+ executorFor(callbackExecutor, ledgerId).execute(
() -> cb.readLacComplete(BKException.Code.IllegalOpException,
ledgerId, null, null, ctx));
}
@Override
public void readEntry(BookieId addr, long ledgerId, long entryId,
ReadEntryCallback cb, Object ctx, int flags, byte[]
masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail, Executor callbackExecutor) {
preReadHook.runHook(addr, ledgerId, entryId)
.thenComposeAsync((res) -> {
LOG.info("[{};L{}] read entry {}", addr, ledgerId,
entryId);
@@ -232,13 +239,13 @@ public class MockBookieClient implements BookieClient {
cb.readEntryComplete(BKException.Code.OK,
ledgerId, entryId, res.slice(), ctx);
}
- }, executor.chooseThread(ledgerId));
+ }, executorFor(callbackExecutor, ledgerId));
}
@Override
public void batchReadEntries(BookieId addr, long ledgerId, long
startEntryId, int maxCount, long maxSize,
BookkeeperInternalCallbacks.BatchedReadEntryCallback cb, Object
ctx, int flags, byte[] masterKey,
- boolean allowFastFail) {
+ boolean allowFastFail, Executor callbackExecutor) {
preBatchReadHook.runHook(addr, ledgerId, startEntryId, maxCount,
maxSize)
.thenComposeAsync((res) -> {
LOG.info("[{};L{}] batch read entries startEntryId:{}
maxCount:{} maxSize:{}",
@@ -267,7 +274,7 @@ public class MockBookieClient implements BookieClient {
cb.readEntriesComplete(BKException.Code.OK,
ledgerId, startEntryId, res, ctx);
}
- }, executor.chooseThread(ledgerId));
+ }, executorFor(callbackExecutor, ledgerId));
}
@Override
@@ -278,8 +285,9 @@ public class MockBookieClient implements BookieClient {
long timeOutInMillis,
boolean piggyBackEntry,
ReadEntryCallback cb,
- Object ctx) {
- executor.executeOrdered(ledgerId,
+ Object ctx,
+ Executor callbackExecutor) {
+ executorFor(callbackExecutor, ledgerId).execute(
() ->
cb.readEntryComplete(BKException.Code.IllegalOpException, ledgerId, entryId,
null, ctx));
}
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestPerChannelBookieClient.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestPerChannelBookieClient.java
index 8c58574840..58325ba463 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestPerChannelBookieClient.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestPerChannelBookieClient.java
@@ -279,7 +279,7 @@ public class TestPerChannelBookieClient extends
BookKeeperClusterTestCase {
}
client.readEntry(1, 1, cb, null,
BookieProtocol.FLAG_DO_FENCING,
- "00000111112222233333".getBytes(), false);
+ "00000111112222233333".getBytes(), false, null);
}
});
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
index 165a8d1cad..f60e47c7ae 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
@@ -22,6 +22,7 @@ package org.apache.bookkeeper.test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
@@ -40,9 +41,12 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -79,6 +83,7 @@ import
org.apache.bookkeeper.test.TestStatsProvider.TestOpStatsLogger;
import org.apache.bookkeeper.test.TestStatsProvider.TestStatsLogger;
import org.apache.bookkeeper.util.ByteBufList;
import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.util.PortManager;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
import org.junit.After;
@@ -916,4 +921,74 @@ public class BookieClientTest {
testDataRefCnfWhenReconnect(false, true, true, false, 10);
testDataRefCnfWhenReconnect(false, true, false, true, 10);
}
+
+ @Test
+ public void testCallbackExecutorV3() throws Exception {
+ testCallbackExecutor(false);
+ }
+
+ @Test
+ public void testCallbackExecutorV2() throws Exception {
+ testCallbackExecutor(true);
+ }
+
+ /**
+ * With a callback executor, responses and connection failures alike are
dispatched on that executor
+ * rather than on the worker thread selected by ledger id.
+ */
+ private void testCallbackExecutor(boolean useV2WireProtocol) throws
Exception {
+ ClientConfiguration conf = new
ClientConfiguration().setUseV2WireProtocol(useV2WireProtocol);
+ BookieClient bc = new BookieClientImpl(conf, eventLoopGroup,
UnpooledByteBufAllocator.DEFAULT,
+ executor, scheduler, NullStatsLogger.INSTANCE,
BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ ExecutorService callbackExecutor = Executors.newSingleThreadExecutor(
+ new DefaultThreadFactory("callback-executor"));
+ try {
+ CompletableFuture<Thread> executorThread = new
CompletableFuture<>();
+ callbackExecutor.execute(() ->
executorThread.complete(Thread.currentThread()));
+ Thread callbackThread = executorThread.get(10, TimeUnit.SECONDS);
+
+ byte[] passwd = new byte[20];
+ Arrays.fill(passwd, (byte) 'a');
+ BookieId addr = bs.getBookieId();
+ DigestManager digestManager = DigestManager.instantiate(1, passwd,
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, useV2WireProtocol);
+ ByteBuf data = Unpooled.buffer(4);
+ data.writeInt(1);
+ ReferenceCounted content =
digestManager.computeDigestAndPackageForSending(1, 0, 4, data,
+ DigestManager.generateMasterKey(passwd),
BookieProtocol.FLAG_NONE);
+
+ CompletableFuture<Thread> addThread = new CompletableFuture<>();
+ bc.addEntry(addr, 1, passwd, 1, content, (rc, ledgerId, entryId,
address, ctx) -> complete(addThread, rc),
+ null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE,
callbackExecutor);
+ assertSame(callbackThread, addThread.get(10, TimeUnit.SECONDS));
+ content.release();
+
+ CompletableFuture<Thread> readThread = new CompletableFuture<>();
+ bc.readEntry(addr, 1, 1, (rc, ledgerId, entryId, buffer, ctx) ->
complete(readThread, rc), null,
+ BookieProtocol.FLAG_NONE, null, false, callbackExecutor);
+ assertSame(callbackThread, readThread.get(10, TimeUnit.SECONDS));
+
+ // Nothing listens on this port: the connection failure completes
the read on the executor too.
+ BookieId unreachable = new BookieSocketAddress("127.0.0.1",
PortManager.nextFreePort()).toBookieId();
+ AtomicInteger failedRc = new AtomicInteger(Code.OK);
+ CompletableFuture<Thread> failedReadThread = new
CompletableFuture<>();
+ bc.readEntry(unreachable, 1, 1, (rc, ledgerId, entryId, buffer,
ctx) -> {
+ failedRc.set(rc);
+ failedReadThread.complete(Thread.currentThread());
+ }, null, BookieProtocol.FLAG_NONE, null, false, callbackExecutor);
+ assertSame(callbackThread, failedReadThread.get(10,
TimeUnit.SECONDS));
+ assertTrue(failedRc.get() != Code.OK);
+ } finally {
+ bc.close();
+ callbackExecutor.shutdown();
+ }
+ }
+
+ private static void complete(CompletableFuture<Thread> future, int rc) {
+ if (rc == Code.OK) {
+ future.complete(Thread.currentThread());
+ } else {
+ future.completeExceptionally(BKException.create(rc));
+ }
+ }
}
