codelipenghui commented on a change in pull request #13533:
URL: https://github.com/apache/pulsar/pull/13533#discussion_r775823434
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
##########
@@ -1729,7 +1729,10 @@ private void getLargestBatchIndexWhenPossible(
ManagedLedgerImpl ml = (ManagedLedgerImpl)
persistentTopic.getManagedLedger();
// If it's not pointing to a valid entry, respond messageId of the
current position.
- if (lastPosition.getEntryId() == -1) {
+ // 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)) {
Review comment:
- compaction horizon < last message -> the compaction cursor does not
reach the end of the original topic, which means not all the data has been
compacted, so we should return the last message ID from the original topic.
- compaction horizon >= last message -> the compaction cursor reach the end
of the original topic, which means all the data been has been compacted, so we
should return the last message ID from the compacted ledger.
--
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]
