codelipenghui commented on code in PR #21248:
URL: https://github.com/apache/pulsar/pull/21248#discussion_r1338063090
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3633,12 +3633,29 @@ public boolean isValidPosition(PositionImpl position) {
log.debug("IsValid position: {} -- last: {}", position, last);
}
- if (position.getEntryId() < 0) {
+ if (!ledgers.containsKey(position.getLedgerId())){
return false;
- } else if (position.getLedgerId() > last.getLedgerId()) {
+ } else if (position.getEntryId() < 0) {
return false;
+ } else if (position.getLedgerId() > last.getLedgerId() &&
position.getLedgerId() != currentLedger.getId()) {
+ // Trim ledgers task will not change the variable
"lastConfirmedEntry", so after trim ledger, LAC might be
+ // smaller than current ledger.
+ // But the largest read position can be set to "{current_ledger:
0}", so if the position's ledger id equals
+ // current ledger, it can larger than LAC. See the logic branch
+ // "else if (position.getLedgerId() == currentLedger.getId()"
+ return false;
+ } else if (position.getLedgerId() == currentLedger.getId() &&
position.getLedgerId() == last.getLedgerId()) {
+ // The current ledger is writing.
+ // The read position can be set to "{LAC + 1}" when subscribe at
LATEST.
+ return position.getEntryId() <= last.getEntryId() + 1;
+ } else if (position.getLedgerId() == currentLedger.getId()) {
+ // The last entry was closed and switch to a new ledger(current
ledger), and current ledger is empty.
+ // The read position can be set to "{current_ledger: 0}".
+ return position.getEntryId() == 0;
} else if (position.getLedgerId() == last.getLedgerId()) {
- return position.getEntryId() <= (last.getEntryId() + 1);
+ // The last entry was closed and switch to a new ledger(current
ledger), and current ledger is empty.
+ // If entry id is larger than LAC, it should be "{current_ledger:
0}".
+ return position.getEntryId() <= last.getEntryId();
Review Comment:
Can it be simplified with
```java
} else if (position.getLedgerId() > currentLedger.getId()) {
return false;
} else if (position.getLedgerId() == currentLedger.getId()) {
// The next position of the active Ledger is a valid read position.
return position.getEntryId() <= last.getEntryId() + 1 ||
position.getEntryId() == 0;
} else if (position.getLedgerId() == last.getLedgerId()) {
return position.getEntryId() <= last.getEntryId();
}
```
--
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]
