[Broker] Optimize ManagedLedger Ledger Ownership Check
@@ -1865,21 +1865,19 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
if (log.isDebugEnabled()) {
log.debug("[{}] Reading entry ledger {}: {}", name, position.getLedgerId(), position.getEntryId());
}
- if (!ledgers.containsKey(position.getLedgerId())) {
- log.error("[{}] Failed to get message with ledger {}:{} the ledgerId does not belong to this topic "
- + "or has been deleted.", name, position.getLedgerId(), position.getEntryId());
- callback.readEntryFailed(new ManagedLedgerException.NonRecoverableLedgerException("Message not found, "
- + "the ledgerId does not belong to this topic or has been deleted"), ctx);
- return;
- }
if (position.getLedgerId() == currentLedger.getId()) {
asyncReadEntry(currentLedger, position, callback, ctx);
- } else {
+ } else if (ledgers.containsKey(position.getLedgerId())) {
getLedgerHandle(position.getLedgerId()).thenAccept(ledger -> asyncReadEntry(ledger, position, callback, ctx)).exceptionally(ex -> {
log.error("[{}] Error opening ledger for reading at position {} - {}", name, position, ex.getMessage());
callback.readEntryFailed(ManagedLedgerException.getManagedLedgerException(ex.getCause()), ctx);
return null;
});
+ } else {
+ log.error("[{}] Failed to get message with ledger {}:{} the ledgerId does not belong to this topic "
+ + "or has been deleted.", name, position.getLedgerId(), position.getEntryId());
+ callback.readEntryFailed(new ManagedLedgerException.NonRecoverableLedgerException("Message not found, "
+ + "the ledgerId does not belong to this topic or has been deleted"), ctx);
}
}
