coderzc commented on code in PR #20718:
URL: https://github.com/apache/pulsar/pull/20718#discussion_r1259677764
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -2070,64 +2070,67 @@ private void getLargestBatchIndexWhenPossible(
// If it's not pointing to a valid entry, respond messageId of the
current position.
// If the compaction cursor reach the end of the topic, respond
messageId from compacted ledger
- Optional<Position> compactionHorizon =
persistentTopic.getCompactedTopic().getCompactionHorizon();
- if (lastPosition.getEntryId() == -1 || (compactionHorizon.isPresent()
- && lastPosition.compareTo((PositionImpl)
compactionHorizon.get()) <= 0)) {
- handleLastMessageIdFromCompactedLedger(persistentTopic, requestId,
partitionIndex,
- markDeletePosition);
- return;
- }
-
- // For a valid position, we read the entry out and parse the batch
size from its metadata.
- CompletableFuture<Entry> entryFuture = new CompletableFuture<>();
- ml.asyncReadEntry(lastPosition, new AsyncCallbacks.ReadEntryCallback()
{
- @Override
- public void readEntryComplete(Entry entry, Object ctx) {
- entryFuture.complete(entry);
+ CompletableFuture<Position> compactionHorizonFuture =
+
persistentTopic.getTopicCompactionService().getLastCompactedPosition();
+
+ compactionHorizonFuture.thenAccept(compactionHorizon -> {
Review Comment:
Fixed.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java:
##########
@@ -1478,25 +1489,14 @@ public synchronized ScheduledExecutorService
getCompactorExecutor() {
return this.compactorExecutor;
}
- // only public so mockito can mock it
- public Compactor newCompactor() throws PulsarServerException {
- return new TwoPhaseCompactor(this.getConfiguration(),
- getClient(), getBookKeeperClient(),
- getCompactorExecutor());
- }
-
- public synchronized Compactor getCompactor() throws PulsarServerException {
- if (this.compactor == null) {
- this.compactor = newCompactor();
- }
- return this.compactor;
- }
-
// This method is used for metrics, which is allowed to as null
// Because it's no operation on the compactor, so let's remove the
synchronized on this method
// to avoid unnecessary lock competition.
public Compactor getNullableCompactor() {
- return this.compactor;
+ if (this.compactionServiceFactory instanceof
PulsarCompactionServiceFactory pulsarCompactedServiceFactory) {
+ return pulsarCompactedServiceFactory.getNullableCompactor();
+ }
Review Comment:
Fixed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
