Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor #21091

Merged
merged 12 commits into from
Sep 4, 2023
Merged

[fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor #21091

Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
adb1b5e
Avoid send batch message to loadbalancer service unit state topic to …
Demogorgon314 Aug 30, 2023
7b7fcb8
Make sure the same batch will batch to same entry.
Demogorgon314 Aug 31, 2023
1292606
Cleanup lastAddedMessageId
Demogorgon314 Aug 31, 2023
9c72c3d
Fix batch
Demogorgon314 Sep 1, 2023
b9ba3f9
Make RawBatchMessageContainerImpl support large batch
Demogorgon314 Sep 1, 2023
97ac2b8
Use MessageIdAdv
Demogorgon314 Sep 1, 2023
88fecc5
use early return
Demogorgon314 Sep 1, 2023
ddf1bb3
Address comments
Demogorgon314 Sep 1, 2023
f2ff5c6
Address comments
Demogorgon314 Sep 1, 2023
d7bac37
Use MessageIdAdv
Demogorgon314 Sep 1, 2023
ffa9303
Update unit test
Demogorgon314 Sep 1, 2023
08571ad
Fix style check
Demogorgon314 Sep 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
use early return
  • Loading branch information
@Demogorgon314
Demogorgon314 committed Sep 1, 2023
commit 88fecc5fe51421b205f2101286e97cf8d4817f33
47 changes: 23 additions & 24 deletions pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -440,32 +440,31 @@ <T> CompletableFuture<Boolean> addToCompactedLedger(

private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
Semaphore outstanding) {
if (batchMessageContainer.getNumMessagesInBatch() <= 0) {
return CompletableFuture.completedFuture(false);
}
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
if (batchMessageContainer.getNumMessagesInBatch() > 0) {
try {
ByteBuf serialized = batchMessageContainer.toByteBuf();
outstanding.acquire();
mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
long start = System.nanoTime();
lh.asyncAddEntry(serialized,
(rc, ledger, eid, ctx) -> {
outstanding.release();
mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
if (rc != BKException.Code.OK) {
bkf.completeExceptionally(BKException.create(rc));
} else {
bkf.complete(true);
}
}, null);
try {
ByteBuf serialized = batchMessageContainer.toByteBuf();
outstanding.acquire();
mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
long start = System.nanoTime();
lh.asyncAddEntry(serialized,
(rc, ledger, eid, ctx) -> {
outstanding.release();
mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
if (rc != BKException.Code.OK) {
bkf.completeExceptionally(BKException.create(rc));
} else {
bkf.complete(true);
}
}, null);

} catch (Throwable t) {
log.error("Failed to add entry", t);
batchMessageContainer.discard((Exception) t);
bkf.completeExceptionally(t);
return bkf;
}
} else {
bkf.complete(false);
} catch (Throwable t) {
log.error("Failed to add entry", t);
batchMessageContainer.discard((Exception) t);
bkf.completeExceptionally(t);
return bkf;
}
return bkf;
}
Expand Down