lhotari commented on code in PR #25177:
URL: https://github.com/apache/pulsar/pull/25177#discussion_r2716656722
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCountEstimator.java:
##########
@@ -183,4 +182,24 @@ static int internalEstimateEntryCountByBytesSize(int
maxEntries, long maxSizeByt
// Ensure at least one entry is always returned as the result
return Math.max((int) Math.min(estimatedEntryCount, maxEntries), 1);
}
+
+ private static Position adjustReadPosition(Position readPosition,
+ NavigableMap<Long,
MLDataFormats.ManagedLedgerInfo.LedgerInfo>
+ ledgersInfo,
+ Long lastLedgerId, long
lastLedgerTotalEntries) {
+ // Adjust the read position to ensure it falls within the valid range
of available ledgers.
+ // This handles special cases such as EARLIEST and LATEST positions by
resetting them
+ // to the first available ledger or the last active ledger,
respectively.
+ if (lastLedgerId != null && readPosition.getLedgerId() >
lastLedgerId.longValue()) {
+ return PositionFactory.create(lastLedgerId,
Math.max(lastLedgerTotalEntries - 1, 0));
+ } else if (lastLedgerId == null && readPosition.getLedgerId() >
ledgersInfo.lastKey()) {
+ Map.Entry<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo>
lastEntry = ledgersInfo.lastEntry();
Review Comment:
addressed
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCountEstimator.java:
##########
@@ -183,4 +182,24 @@ static int internalEstimateEntryCountByBytesSize(int
maxEntries, long maxSizeByt
// Ensure at least one entry is always returned as the result
return Math.max((int) Math.min(estimatedEntryCount, maxEntries), 1);
}
+
+ private static Position adjustReadPosition(Position readPosition,
+ NavigableMap<Long,
MLDataFormats.ManagedLedgerInfo.LedgerInfo>
+ ledgersInfo,
+ Long lastLedgerId, long
lastLedgerTotalEntries) {
+ // Adjust the read position to ensure it falls within the valid range
of available ledgers.
+ // This handles special cases such as EARLIEST and LATEST positions by
resetting them
+ // to the first available ledger or the last active ledger,
respectively.
+ if (lastLedgerId != null && readPosition.getLedgerId() >
lastLedgerId.longValue()) {
+ return PositionFactory.create(lastLedgerId,
Math.max(lastLedgerTotalEntries - 1, 0));
+ } else if (lastLedgerId == null && readPosition.getLedgerId() >
ledgersInfo.lastKey()) {
+ Map.Entry<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo>
lastEntry = ledgersInfo.lastEntry();
+ if (lastEntry != null) {
+ return PositionFactory.create(lastEntry.getKey(),
Math.max(lastEntry.getValue().getEntries() - 1, 0));
+ }
+ } else if (readPosition.getLedgerId() < ledgersInfo.firstKey()) {
+ return PositionFactory.create(ledgersInfo.firstKey(), 0);
Review Comment:
addressed
--
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]
