You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/01/22 08:27:14 UTC
[PR] [fix][ml] Fix NoSuchElementException in EntryCountEstimator caused by a race condition [pulsar]
lhotari opened a new pull request, #25177:
URL: https://github.com/apache/pulsar/pull/25177
Fixes #25174
### Motivation
An NoSuchElementException could occur in EntryCountEstimator, causing dispatching to stop.
This condition could happen as a result of a race condition caused by ledger trimming. The ledgersInfo map appear to be be empty.
### Modifications
- check if ledgersInfo map is empty
- catch NoSuchElementException when firstKey or lastKey is accessed to avoid race conditions after checking if the map was empty
- add tests
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix NoSuchElementException in EntryCountEstimator caused by a race condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix NoSuchElementException in EntryCountEstimator caused by a race condition [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #25177:
URL: https://github.com/apache/pulsar/pull/25177
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Fix NoSuchElementException in EntryCountEstimator caused by a race condition [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on code in PR #25177:
URL: https://github.com/apache/pulsar/pull/25177#discussion_r2716479246
##########
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:
The method call `ledgersInfo.lastEntry()` and `ledgersInfo.lastKey()` in the previous line may return different ledgers due to race condition.
##########
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:
```
var lastEntry = ledgersInfo.lastEntry();
...
} else if (lastLedgerId == null && lastEntry != null && readPosition.getLedgerId() > lastEntry.getKey()) {
~~ Map.Entry<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo> lastEntry = ledgersInfo.lastEntry();~~
```
##########
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:
```java
var firstEntry = ledgersInfo.firstEntry();
...
} else if (firstEntry != null && readPosition.getLedgerId() < firstEntry.getKey()) {
return PositionFactory.create(firstEntry.getKey(), 0);
```
##########
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:
Same here, those two `ledgersInfo.firstKey()` method calls may return different ledger, I think we can reuse the method call result in a local variable and switch to safe alternate `firstEntry()` and `lastEntry()` with nullness check.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org