You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Denovo1998 (via GitHub)" <gi...@apache.org> on 2026/10/14 13:40:27 UTC
[PR] [fix][ml] Enforce the add-entry handover batch byte limit [pulsar]
Denovo1998 opened a new pull request, #26901:
URL: https://github.com/apache/pulsar/pull/26901
### Motivation
The add-entry handover batching introduced in #26717 checks the accumulated weight before taking the next task, allowing a batch to exceed `managedLedgerAddEntryHandoverMaxBatchBytesSize`. With a limit of 10, entries weighing `[4, 4, 4]` run in one batch, and `[4, 100]` also share a batch instead of giving the oversized entry its own batch. This delays other work sharing the ledger executor beyond the configured byte budget.
### Modifications
- Check the next task's weight before removing it from the queue. Leave an entry that does not fit at the queue head for the next batch, preserving FIFO order.
- Run an oversized entry in a batch of its own, while preserving item-only batching when the byte limit is disabled.
- Update the configuration descriptions to document the byte bound and oversized-entry behavior.
- Extend the existing wrapper tests and add deterministic regressions through `ManagedLedgerImpl.asyncAddEntry` with real ZooKeeper and BookKeeper fixtures.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- `ManagedLedgerBkTest.testAddEntryHandoverYieldsBeforeExceedingByteLimit` covers `[4, 4, 4]`, `[4, 100, 4]`, `[100, 4]`, and an exact-limit batch. It verifies that competing executor work runs between batches and that entries retain their order and contents.
- `BatchingExecutorWrapperTest` covers strict weight boundaries, oversized tasks, disabled weight limits, and failures while reading a task's weight.
- Before the fix, two of the four real-BookKeeper regression cases failed deterministically. On base `be41491e7d9e`, all 34 targeted test cases and `quickCheck` passed with the patch.
```bash
./gradlew :managed-ledger:test \
--tests 'BatchingExecutorWrapperTest' \
--tests 'ManagedLedgerBkTest.testAddEntryHandoverYieldsBeforeExceedingByteLimit' \
--tests 'ManagedLedgerTest.test*AddEntryHandover*' \
--tests 'ManagedLedgerTest.testRejectedAddEntryReleasesTheRetainedBuffer' \
-PtestRetryCount=0 -PtestFailFast=false -PtestMaxParallelForks=1 --max-workers=2
./gradlew quickCheck --max-workers=1
```
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [x] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
Threading model: handover batches yield before taking an entry that would exceed the byte budget. Entries continue to run on the same ordered ledger executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Enforce the add-entry handover batch byte limit [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #26901:
URL: https://github.com/apache/pulsar/pull/26901
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Enforce the add-entry handover batch byte limit [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26901:
URL: https://github.com/apache/pulsar/pull/26901#discussion_r4279146522
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java:
##########
@@ -211,13 +211,30 @@ private void runHandoverBatch() {
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) {
Review Comment:
Optional, nonblocking: consider adding a concurrent case with task weight 2 and byte limit 3. Existing weighted concurrency coverage uses weight 1 and limit 3, reaching the limit exactly, so it does not exercise leaving a non-fitting head queued across batches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Enforce the add-entry handover batch byte limit [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26901:
URL: https://github.com/apache/pulsar/pull/26901#discussion_r4283247465
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/BatchingExecutorWrapper.java:
##########
@@ -211,13 +211,30 @@ private void runHandoverBatch() {
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) {
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org