You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/15 22:57:44 UTC
(pulsar) branch master updated: [fix][ml] Enforce the add-entry handover batch byte limit (#26901)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new c806da5aeb4b [fix][ml] Enforce the add-entry handover batch byte limit (#26901)
c806da5aeb4b is described below
commit c806da5aeb4bcb622e758688bd81a8b2030f0baa
Author: sinan liu <li...@gmail.com>
AuthorDate: Fri Oct 16 06:57:37 2026 +0800
[fix][ml] Enforce the add-entry handover batch byte limit (#26901)
---
conf/broker.conf | 10 +--
conf/standalone.conf | 10 +--
.../bookkeeper/mledger/ManagedLedgerConfig.java | 16 ++--
.../mledger/impl/BatchingExecutorWrapper.java | 35 ++++++---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 10 +--
.../mledger/impl/BatchingExecutorWrapperTest.java | 69 +++++++++++++++---
.../mledger/impl/ManagedLedgerBkTest.java | 85 ++++++++++++++++++++++
.../apache/pulsar/broker/ServiceConfiguration.java | 10 +--
8 files changed, 199 insertions(+), 46 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 46e8b99a2b6e..b9f268a4dfcd 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1561,8 +1561,8 @@ managedLedgerReadEntriesCallbackInline=true
# Maximum number of add entry requests handed over to the managed ledger's executor thread in one batch.
# Publishing threads queue adds for the ledger's executor, which takes them over in batches of up to this
-# many adds and processes each batch before other tasks on that thread can run. A batch also stops taking
-# adds once their entries add up to managedLedgerAddEntryHandoverMaxBatchBytesSize bytes. A larger value reduces scheduling
+# many adds and processes each batch before other tasks on that thread can run. A batch also stops before taking
+# an entry that would exceed managedLedgerAddEntryHandoverMaxBatchBytesSize bytes. A larger value reduces scheduling
# overhead and contention between publishing threads under high publish rates, but keeps the executor thread
# occupied for longer per batch, which can delay add completions, reads and cursor notifications for the
# ledgers that share the thread. A smaller value favors those tasks over add throughput. Set to 0 or 1 to
@@ -1571,11 +1571,11 @@ managedLedgerReadEntriesCallbackInline=true
# already open keep the value they opened with.
managedLedgerAddEntryHandoverMaxBatchItems=1024
-# Total size in bytes of the entries after which a batch of add entry requests handed over to the managed
-# ledger's executor thread stops taking more. This keeps a ledger with large entries from occupying the
+# Maximum total entry size in bytes of a batch of add entry requests handed over to the managed
+# ledger's executor thread. This keeps a ledger with large entries from occupying the
# executor thread for as long as a full batch of managedLedgerAddEntryHandoverMaxBatchItems adds would,
# which would delay add completions, reads and cursor notifications for the ledgers that share the thread.
-# A batch always takes at least one add, even one whose entry is larger than this. Set to 0 to limit
+# An entry larger than this runs in a batch of its own. Set to 0 to limit
# batches only by their number of adds. The default is 5 MB.
# This setting is dynamic: updates apply to managed ledgers opened after the change; ledgers that are
# already open keep the value they opened with.
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 2cc05bd58788..4e48a012d182 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -1055,8 +1055,8 @@ managedLedgerReadEntriesCallbackInline=true
# Maximum number of add entry requests handed over to the managed ledger's executor thread in one batch.
# Publishing threads queue adds for the ledger's executor, which takes them over in batches of up to this
-# many adds and processes each batch before other tasks on that thread can run. A batch also stops taking
-# adds once their entries add up to managedLedgerAddEntryHandoverMaxBatchBytesSize bytes. A larger value reduces scheduling
+# many adds and processes each batch before other tasks on that thread can run. A batch also stops before taking
+# an entry that would exceed managedLedgerAddEntryHandoverMaxBatchBytesSize bytes. A larger value reduces scheduling
# overhead and contention between publishing threads under high publish rates, but keeps the executor thread
# occupied for longer per batch, which can delay add completions, reads and cursor notifications for the
# ledgers that share the thread. A smaller value favors those tasks over add throughput. Set to 0 or 1 to
@@ -1065,11 +1065,11 @@ managedLedgerReadEntriesCallbackInline=true
# already open keep the value they opened with.
managedLedgerAddEntryHandoverMaxBatchItems=1024
-# Total size in bytes of the entries after which a batch of add entry requests handed over to the managed
-# ledger's executor thread stops taking more. This keeps a ledger with large entries from occupying the
+# Maximum total entry size in bytes of a batch of add entry requests handed over to the managed
+# ledger's executor thread. This keeps a ledger with large entries from occupying the
# executor thread for as long as a full batch of managedLedgerAddEntryHandoverMaxBatchItems adds would,
# which would delay add completions, reads and cursor notifications for the ledgers that share the thread.
-# A batch always takes at least one add, even one whose entry is larger than this. Set to 0 to limit
+# An entry larger than this runs in a batch of its own. Set to 0 to limit
# batches only by their number of adds. The default is 5 MB.
# This setting is dynamic: updates apply to managed ledgers opened after the change; ledgers that are
# already open keep the value they opened with.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index a9beecba0cc6..a32053fb0874 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -472,8 +472,8 @@ public class ManagedLedgerConfig {
* Set the maximum number of add entry requests handed over to the ledger's executor thread in one batch.
*
* <p>Publishing threads queue adds for the ledger's executor, which takes them over in batches of up to this many
- * adds and processes each batch before other tasks on that thread can run. A batch also stops taking adds once
- * their entries add up to {@link #setAddEntryHandoverMaxBatchBytesSize(long)} bytes. A larger value reduces
+ * adds and processes each batch before other tasks on that thread can run. A batch also stops before taking an
+ * entry that would exceed {@link #setAddEntryHandoverMaxBatchBytesSize(long)} bytes. A larger value reduces
* scheduling overhead and contention between publishing threads, but keeps the executor thread occupied for longer
* per batch, which can delay add completions, reads and cursor notifications for the ledgers that share the
* thread. 0 or 1 disables batching: each add is handed over to the executor as a task of its own.
@@ -492,8 +492,8 @@ public class ManagedLedgerConfig {
}
/**
- * The total size in bytes of the entries after which a batch of add entry requests handed over to the ledger's
- * executor thread stops taking more, or 0 when batches are limited only by their number of adds.
+ * The maximum total entry size in bytes of a batch of add entry requests handed over to the ledger's executor
+ * thread, or 0 when batches are limited only by their number of adds. An oversized entry runs in its own batch.
*
* @see #setAddEntryHandoverMaxBatchBytesSize(long)
*/
@@ -502,13 +502,13 @@ public class ManagedLedgerConfig {
}
/**
- * Set the total size in bytes of the entries after which a batch of add entry requests handed over to the ledger's
- * executor thread stops taking more.
+ * Set the maximum total entry size in bytes of a batch of add entry requests handed over to the ledger's executor
+ * thread.
*
* <p>This keeps a ledger with large entries from occupying the executor thread for as long as a full batch of
* {@link #setAddEntryHandoverMaxBatchItems(int)} adds would, which would delay add completions, reads and cursor
- * notifications for the ledgers that share the thread. A batch always takes at least one add, even one whose entry
- * is larger than this. 0 removes the limit, so that batches are limited only by their number of adds. It has no
+ * notifications for the ledgers that share the thread. An entry larger than this runs in a batch of its own.
+ * 0 removes the limit, so that batches are limited only by their number of adds. It has no
* effect when batching is disabled.
*
* <p>The value is captured when the ledger is opened; subsequent changes, including
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java
index 5e6bdb9a99bc..54e964c47e6a 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java
@@ -33,8 +33,8 @@ import org.jspecify.annotations.Nullable;
* An {@link Executor} that hands tasks over to a delegate executor in batches instead of one task at a time.
*
* <p>Submitting threads append tasks to a multi-producer single-consumer queue. The thread that finds no handover batch
- * scheduled submits one task to the delegate, and that task runs the queued tasks until it has run {@code maxItems}
- * tasks or their weights add up to {@code maxWeight}. The delegate's own queue then sees one task per batch rather than
+ * scheduled submits one task to the delegate, and that task runs up to {@code maxItems} queued tasks whose total
+ * weight does not exceed {@code maxWeight}. The delegate's own queue then sees one task per batch rather than
* one per submitted task, so concurrent submitters contend on it once per batch, and other work on the delegate does
* not wait behind a task per submission. When more tasks remain after a batch, the next batch is submitted to the
* delegate as a new task, so that tasks submitted to the delegate directly can run in between.
@@ -42,7 +42,7 @@ import org.jspecify.annotations.Nullable;
* <p>The weight limit keeps a batch of heavy tasks from occupying the delegate for as long as {@code maxItems} of
* them would, so that the other work sharing the delegate is not starved. A task that implements
* {@link WeightedRunnable} weighs {@link WeightedRunnable#getWeight()}; any other task weighs 0 and is limited only by
- * {@code maxItems}. A batch always runs at least one task, even one that weighs more than {@code maxWeight}.
+ * {@code maxItems}. A task that weighs more than {@code maxWeight} runs in a batch of its own.
*
* <p>Tasks submitted by one thread run in the order that thread submitted them. The delegate must run the tasks
* submitted to it one at a time, such as a single-threaded executor, since the queue supports a single consumer.
@@ -91,8 +91,8 @@ class BatchingExecutorWrapper implements Executor {
* @param delegate the executor that runs the handover batches; it must run its tasks one at a time
* @param queueChunkSize the chunk size of the handover queue, which grows by linking chunks of this size
* @param maxItems the maximum number of tasks run by one handover batch; must be greater than 1
- * @param maxWeight the total weight of tasks after which a handover batch stops taking more tasks; must be
- * positive, and {@link Long#MAX_VALUE} leaves batches limited only by {@code maxItems}
+ * @param maxWeight the maximum total weight of a handover batch, except for a single task heavier than this;
+ * must be positive, and {@link Long#MAX_VALUE} leaves batches limited only by {@code maxItems}
* @param runFailureConsumer receives what a task run by a handover batch, or the rejected task handler, throws,
* so that the remaining tasks still run
* @param rejectedTaskHandler receives each queued task that will not run because the delegate rejected its
@@ -211,13 +211,30 @@ class BatchingExecutorWrapper implements Executor {
MpscUnboundedArrayQueue<Runnable> queue = handoverQueue;
int items = 0;
long weight = 0;
+ boolean weightLimited = maxWeight != Long.MAX_VALUE;
Runnable command;
- while (items < maxItems && weight < maxWeight && (command = queue.relaxedPoll()) != null) {
+ while (items < maxItems && (!weightLimited || weight < maxWeight)
+ && (command = queue.relaxedPeek()) != null) {
+ long commandWeight;
+ try {
+ commandWeight = command instanceof WeightedRunnable weightedCommand ? weightedCommand.getWeight() : 0;
+ } catch (Throwable t) {
+ queue.relaxedPoll();
+ items++;
+ runFailureConsumer.accept(t);
+ continue;
+ }
+ if (weightLimited && items > 0 && commandWeight > maxWeight - weight) {
+ // Leave the head queued for the next batch: moving it to the tail would reorder tasks.
+ break;
+ }
+ // The handover flag makes this the sole consumer, so the task seen by peek is still at the head.
+ queue.relaxedPoll();
items++;
+ if (weightLimited) {
+ weight += commandWeight;
+ }
try {
- if (command instanceof WeightedRunnable weightedCommand) {
- weight += weightedCommand.getWeight();
- }
command.run();
} catch (Throwable t) {
runFailureConsumer.accept(t);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 98394275779d..14fb0ec3a239 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -380,11 +380,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// thread that finds no handover batch scheduled submits one, and that task runs the adds queued by then. The
// executor's own queue then sees one task per batch rather than one per add, so concurrent publishers contend on
// it once per batch, and other executor work (add completions, cursor notifications) does not wait behind a task
- // per published message. A handover batch runs at most addEntryHandoverMaxBatchItems adds, and stops taking more
- // once their entries add up to addEntryHandoverMaxBatchBytesSize bytes, so that a ledger with large entries does
- // not hold the executor thread from the ledgers that share it for long. Both are captured when the ledger is
- // opened; a max batch items of 0 or 1 disables batching, and each add is then handed over to the executor as a
- // task of its own.
+ // per published message. A handover batch runs at most addEntryHandoverMaxBatchItems adds, and stops before taking
+ // an entry that would exceed addEntryHandoverMaxBatchBytesSize bytes (an oversized entry runs alone), so a ledger
+ // with large entries does not hold the executor thread from the ledgers that share it for long. Both are captured
+ // when the ledger is opened; a max batch items of 0 or 1 disables batching, and each add is then handed over to
+ // the executor as a task of its own.
private final Executor addEntryBatchingExecutor;
// Chunk size of the add entry handover queue; the queue grows by linking chunks of this size when a batch backs up.
private static final int ADD_ENTRY_HANDOVER_QUEUE_CHUNK_SIZE = 512;
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapperTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapperTest.java
index f897f8a061c8..62b2b54aae86 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapperTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapperTest.java
@@ -118,7 +118,7 @@ public class BatchingExecutorWrapperTest {
}
@Test
- public void testBatchStopsTakingTasksOnceTheirWeightsReachMaxWeight() {
+ public void testBatchStopsBeforeExceedingMaxWeight() {
ManualExecutor delegate = new ManualExecutor();
BatchingExecutorWrapper wrapper = newWrapper(delegate, 1024, 10);
List<Integer> ran = new ArrayList<>();
@@ -129,10 +129,11 @@ public class BatchingExecutorWrapperTest {
delegate.runNext();
- // 4 + 4 is below the limit, so the batch takes a third task; 12 reaches it, so the batch stops there.
- assertThat(ran).containsExactly(0, 1, 2);
+ assertThat(ran).containsExactly(0, 1);
assertThat(delegate.tasks).hasSize(1);
delegate.runNext();
+ assertThat(ran).containsExactly(0, 1, 2, 3);
+ delegate.runNext();
assertThat(ran).containsExactly(0, 1, 2, 3, 4, 5);
}
@@ -141,14 +142,58 @@ public class BatchingExecutorWrapperTest {
ManualExecutor delegate = new ManualExecutor();
BatchingExecutorWrapper wrapper = newWrapper(delegate, 1024, 10);
List<Integer> ran = new ArrayList<>();
- wrapper.execute(weighted(100, () -> ran.add(0)));
- wrapper.execute(weighted(1, () -> ran.add(1)));
+ wrapper.execute(weighted(4, () -> ran.add(0)));
+ wrapper.execute(weighted(100, () -> ran.add(1)));
+ wrapper.execute(weighted(4, () -> ran.add(2)));
delegate.runNext();
assertThat(ran).containsExactly(0);
delegate.runNext();
assertThat(ran).containsExactly(0, 1);
+ delegate.runNext();
+ assertThat(ran).containsExactly(0, 1, 2);
+ }
+
+ @Test
+ public void testNoWeightLimitDoesNotOverflow() {
+ ManualExecutor delegate = new ManualExecutor();
+ BatchingExecutorWrapper wrapper = newWrapper(delegate, 1024, NO_WEIGHT_LIMIT);
+ List<Integer> ran = new ArrayList<>();
+ wrapper.execute(weighted(Long.MAX_VALUE, () -> ran.add(0)));
+ wrapper.execute(weighted(1, () -> ran.add(1)));
+
+ delegate.runNext();
+
+ assertThat(ran).containsExactly(0, 1);
+ assertThat(delegate.tasks).isEmpty();
+ }
+
+ @Test
+ public void testTaskWeightFailureIsProcessedAndTheBatchContinues() {
+ ManualExecutor delegate = new ManualExecutor();
+ List<Throwable> failures = new ArrayList<>();
+ BatchingExecutorWrapper wrapper = new BatchingExecutorWrapper(delegate, QUEUE_CHUNK_SIZE, 1024, 10,
+ failures::add, FAIL_ON_REJECTED_TASK);
+ RuntimeException failure = new RuntimeException("weight failed");
+ List<Integer> ran = new ArrayList<>();
+ wrapper.execute(new BatchingExecutorWrapper.WeightedRunnable() {
+ @Override
+ public long getWeight() {
+ throw failure;
+ }
+
+ @Override
+ public void run() {
+ ran.add(0);
+ }
+ });
+ wrapper.execute(weighted(4, () -> ran.add(1)));
+
+ delegate.runAll();
+
+ assertThat(failures).containsExactly(failure);
+ assertThat(ran).containsExactly(1);
}
@Test
@@ -420,11 +465,17 @@ public class BatchingExecutorWrapperTest {
@DataProvider
public Object[][] batchLimits() {
- return new Object[][] {{2, NO_WEIGHT_LIMIT}, {1024, NO_WEIGHT_LIMIT}, {1024, 3L}};
+ return new Object[][] {
+ {2, NO_WEIGHT_LIMIT, 1L},
+ {1024, NO_WEIGHT_LIMIT, 1L},
+ {1024, 3L, 1L},
+ {1024, 3L, 2L}
+ };
}
@Test(timeOut = 30000, dataProvider = "batchLimits")
- public void testConcurrentSubmittersKeepPerThreadOrder(int maxItems, long maxWeight) throws Exception {
+ public void testConcurrentSubmittersKeepPerThreadOrder(int maxItems, long maxWeight, long taskWeight)
+ throws Exception {
int threads = 8;
int tasksPerThread = 10000;
ExecutorService delegate = Executors.newSingleThreadExecutor();
@@ -445,7 +496,7 @@ public class BatchingExecutorWrapperTest {
start.await();
for (int i = 0; i < tasksPerThread; i++) {
int task = i;
- wrapper.execute(weighted(1, () -> {
+ wrapper.execute(weighted(taskWeight, () -> {
ran.add(task);
completed.countDown();
}));
@@ -463,7 +514,7 @@ public class BatchingExecutorWrapperTest {
assertThat(failures).isEmpty();
// Tasks from one thread run in the order that thread submitted them.
for (List<Integer> ran : ranByThread) {
- assertThat(ran).hasSize(tasksPerThread).isSorted();
+ assertThat(ran).hasSize(tasksPerThread).isSorted().doesNotHaveDuplicates();
}
}
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
index 5274b9b868c2..c3f6355d56d6 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
@@ -30,6 +30,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
@@ -88,6 +89,90 @@ public class ManagedLedgerBkTest extends BookKeeperClusterTestCase {
baseClientConf.setUseV2WireProtocol(true);
}
+ @DataProvider
+ public Object[][] addEntryHandoverByteLimits() {
+ return new Object[][] {
+ {new int[] {4, 4, 4}, 2},
+ {new int[] {4, 100, 4}, 1},
+ {new int[] {100, 4}, 1},
+ {new int[] {4, 6, 4}, 2}
+ };
+ }
+
+ @Test(timeOut = 30000, dataProvider = "addEntryHandoverByteLimits")
+ public void testAddEntryHandoverYieldsBeforeExceedingByteLimit(int[] entrySizes, int firstBatchEntries)
+ throws Exception {
+ @Cleanup("shutdown")
+ ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerConfig config = defaultConfig()
+ .setEnsembleSize(2).setWriteQuorumSize(2).setAckQuorumSize(2)
+ .setAddEntryHandoverMaxBatchBytesSize(10);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("handover-byte-limit-" + UUID.randomUUID(), config);
+ ManagedCursor cursor = ledger.openCursor("c1");
+ CountDownLatch workerStarted = new CountDownLatch(1);
+ CountDownLatch resumeWorker = new CountDownLatch(1);
+ CompletableFuture<Long> entriesBeforeOtherWork = new CompletableFuture<>();
+ List<CompletableFuture<Position>> positions = new ArrayList<>();
+ try {
+ // Hold the real BookKeeper worker until all adds and the competing task have been queued.
+ ledger.getExecutor().execute(() -> {
+ workerStarted.countDown();
+ try {
+ if (!resumeWorker.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting to resume the BookKeeper worker");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ });
+ assertThat(workerStarted.await(10, TimeUnit.SECONDS)).isTrue();
+ for (int i = 0; i < entrySizes.length; i++) {
+ CompletableFuture<Position> position = new CompletableFuture<>();
+ positions.add(position);
+ ByteBuf data = Unpooled.buffer(entrySizes[i]).writeZero(entrySizes[i]);
+ data.setByte(0, i);
+ try {
+ ledger.asyncAddEntry(data, new AddEntryCallback() {
+ @Override
+ public void addComplete(Position addedPosition, ByteBuf entryData, Object ctx) {
+ position.complete(addedPosition);
+ }
+
+ @Override
+ public void addFailed(ManagedLedgerException exception, Object ctx) {
+ position.completeExceptionally(exception);
+ }
+ }, null);
+ } finally {
+ data.release();
+ }
+ }
+ // This task must run between the first batch and the adds that do not fit in it.
+ ledger.getExecutor().execute(() -> entriesBeforeOtherWork.complete(ledger.getCurrentLedgerEntries()));
+ } finally {
+ resumeWorker.countDown();
+ }
+
+ assertThat(entriesBeforeOtherWork.get(10, TimeUnit.SECONDS)).isEqualTo(firstBatchEntries);
+ List<Position> added = new ArrayList<>();
+ for (CompletableFuture<Position> position : positions) {
+ added.add(position.get(10, TimeUnit.SECONDS));
+ }
+ assertThat(added).isSorted().doesNotHaveDuplicates();
+ List<Entry> entries = cursor.readEntries(entrySizes.length);
+ try {
+ assertThat(entries).hasSize(entrySizes.length);
+ for (int i = 0; i < entrySizes.length; i++) {
+ assertThat(entries.get(i).getLength()).isEqualTo(entrySizes[i]);
+ assertThat(entries.get(i).getDataBuffer().getByte(0)).isEqualTo((byte) i);
+ assertThat(entries.get(i).getPosition()).isEqualTo(added.get(i));
+ }
+ } finally {
+ entries.forEach(Entry::release);
+ }
+ }
+
@Test
public void testBatchReadNotUsedWithV3WireProtocolClient() throws Exception {
// The suite's client uses the v2 wire protocol; a v3 client cannot batch read, so reads must not use it
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index d705a901a532..1db070a76dff 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2755,7 +2755,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
doc = "Maximum number of add entry requests handed over to the managed ledger's executor thread in one "
+ "batch. Publishing threads queue adds for the ledger's executor, which takes them over in "
+ "batches of up to this many adds and processes each batch before other tasks on that thread can "
- + "run. A batch also stops taking adds once their entries add up to "
+ + "run. A batch also stops before taking an entry that would exceed "
+ "managedLedgerAddEntryHandoverMaxBatchBytesSize bytes. "
+ "A larger value reduces scheduling overhead and contention between publishing threads under "
+ "high publish rates, but keeps the executor thread occupied for longer per batch, which can "
@@ -2767,12 +2767,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(category = CATEGORY_STORAGE_ML,
dynamic = true,
- doc = "Total size in bytes of the entries after which a batch of add entry requests handed over to the "
- + "managed ledger's executor thread stops taking more. This keeps a ledger with large entries "
+ doc = "Maximum total entry size in bytes of a batch of add entry requests handed over to the "
+ + "managed ledger's executor thread. This keeps a ledger with large entries "
+ "from occupying the executor thread for as long as a full batch of "
+ "managedLedgerAddEntryHandoverMaxBatchItems adds would, which would delay add completions, "
- + "reads and cursor notifications for the ledgers that share the thread. A batch always takes at "
- + "least one add, even one whose entry is larger than this. Set to 0 to limit batches only by "
+ + "reads and cursor notifications for the ledgers that share the thread. An entry larger than "
+ + "this runs in a batch of its own. Set to 0 to limit batches only by "
+ "their number of adds. Updates apply to managed ledgers opened after the change; ledgers that "
+ "are already open keep the value they opened with.")
private long managedLedgerAddEntryHandoverMaxBatchBytesSize = 5 * 1024 * 1024;