This is an automated email from the ASF dual-hosted git repository. merlimat pushed a commit to branch branch-4.18 in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit f77d07c1b08de6d554e4f45702a06589f30c5a93 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)); + } + } }
