You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/03/30 10:44:02 UTC
[GitHub] [pulsar] wuzhanpeng opened a new pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
wuzhanpeng opened a new pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087
This pull request resolves #10086
## Motivation
Although #7111 and #9136 solved the problems of
1. current ledger is full and cannot be rolled
2. cursor has subscribed to an expired ledger, which makes the cleaning thread unable to clean up
respectively. However, when we close and open a managed-ledger(such like brokers shutdown unexpectedly or topics unload due to rebalance), the managed-ledger will create an empty ledger after initialization causing the current cleanup logic failed to take effect. Therefore I think we need a more unified entrance to solve the cursor update problem, and then further solve the problem of clearing expired ledgers.
## Expected Behavior
1. Able to cleanup expired data after managed-ledger re-open
2. A more unified entrance to implement cursor update
## Modification
1. move the cursor update logic to `internalTrimConsumedLedgers` from `ManagedLedgerImpl`'s callback `createComplete`
2. update `lastConfirmedEntry` if necessary when updating cursors
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087#issuecomment-845838027
@merlimat @eolivelli Please help review this PR thanks
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087#discussion_r607851452
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -2140,6 +2133,42 @@ public void addWaitingEntryCallBack(WaitingEntryCallBack cb) {
this.waitingEntryCallBacks.add(cb);
}
+ public void maybeUpdateCursorBeforeTrimmingConsumedLedger() {
+ for (ManagedCursor cursor : cursors) {
+ PositionImpl lastAckedPosition = (PositionImpl) cursor.getMarkDeletedPosition();
+ LedgerInfo currPointedLedger = ledgers.get(lastAckedPosition.getLedgerId());
+ LedgerInfo nextPointedLedger = Optional.ofNullable(ledgers.higherEntry(lastAckedPosition.getLedgerId()))
+ .map(Map.Entry::getValue).orElse(null);
+
+ if (currPointedLedger != null) {
+ if (nextPointedLedger != null) {
+ if (lastAckedPosition.getEntryId() != -1 &&
+ lastAckedPosition.getEntryId() + 1 >= currPointedLedger.getEntries()) {
+ lastAckedPosition = new PositionImpl(nextPointedLedger.getLedgerId(), -1);
+ }
+ } else {
+ log.debug("No need to reset cursor: {}, current ledger is the last ledger.", cursor);
+ }
+ } else {
+ log.warn("Cursor: {} does not exist in the managed-ledger.", cursor);
+ }
+
+ if (!lastAckedPosition.equals((PositionImpl) cursor.getMarkDeletedPosition())) {
+ try {
+ log.info("Reset cursor:{} to {} since ledger consumed completely", cursor, lastAckedPosition);
+ if (lastConfirmedEntry.compareTo(lastAckedPosition) < 0) {
+ lastConfirmedEntry = lastAckedPosition;
Review comment:
This will break the behavior of the `lastConfirmedEntry` because we will get such as (10, -1) when get the last confirm entry.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] wuzhanpeng commented on a change in pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
Posted by GitBox <gi...@apache.org>.
wuzhanpeng commented on a change in pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087#discussion_r608441424
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -2140,6 +2133,42 @@ public void addWaitingEntryCallBack(WaitingEntryCallBack cb) {
this.waitingEntryCallBacks.add(cb);
}
+ public void maybeUpdateCursorBeforeTrimmingConsumedLedger() {
+ for (ManagedCursor cursor : cursors) {
+ PositionImpl lastAckedPosition = (PositionImpl) cursor.getMarkDeletedPosition();
+ LedgerInfo currPointedLedger = ledgers.get(lastAckedPosition.getLedgerId());
+ LedgerInfo nextPointedLedger = Optional.ofNullable(ledgers.higherEntry(lastAckedPosition.getLedgerId()))
+ .map(Map.Entry::getValue).orElse(null);
+
+ if (currPointedLedger != null) {
+ if (nextPointedLedger != null) {
+ if (lastAckedPosition.getEntryId() != -1 &&
+ lastAckedPosition.getEntryId() + 1 >= currPointedLedger.getEntries()) {
+ lastAckedPosition = new PositionImpl(nextPointedLedger.getLedgerId(), -1);
+ }
+ } else {
+ log.debug("No need to reset cursor: {}, current ledger is the last ledger.", cursor);
+ }
+ } else {
+ log.warn("Cursor: {} does not exist in the managed-ledger.", cursor);
+ }
+
+ if (!lastAckedPosition.equals((PositionImpl) cursor.getMarkDeletedPosition())) {
+ try {
+ log.info("Reset cursor:{} to {} since ledger consumed completely", cursor, lastAckedPosition);
+ if (lastConfirmedEntry.compareTo(lastAckedPosition) < 0) {
+ lastConfirmedEntry = lastAckedPosition;
Review comment:
Yes, I have made a reminder in the place of **Modification** above. If we don’t modify `lastConfirmedEntry` here, the `updateCursor` can’t really succeed and exit with message `[{}] Failed mark delete due to invalid markDelete {} is ahead of last-confirmed-entry {} for cursor [{}]`. Then you will find that the state of cursor in memory is inconsistent with the metadata. This scenario may become difficult to understand.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #10087: [BUGFIX] cannot cleanup expired data after managed-ledger restart
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #10087:
URL: https://github.com/apache/pulsar/pull/10087#discussion_r609185476
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -2140,6 +2133,42 @@ public void addWaitingEntryCallBack(WaitingEntryCallBack cb) {
this.waitingEntryCallBacks.add(cb);
}
+ public void maybeUpdateCursorBeforeTrimmingConsumedLedger() {
+ for (ManagedCursor cursor : cursors) {
+ PositionImpl lastAckedPosition = (PositionImpl) cursor.getMarkDeletedPosition();
+ LedgerInfo currPointedLedger = ledgers.get(lastAckedPosition.getLedgerId());
+ LedgerInfo nextPointedLedger = Optional.ofNullable(ledgers.higherEntry(lastAckedPosition.getLedgerId()))
+ .map(Map.Entry::getValue).orElse(null);
+
+ if (currPointedLedger != null) {
+ if (nextPointedLedger != null) {
+ if (lastAckedPosition.getEntryId() != -1 &&
+ lastAckedPosition.getEntryId() + 1 >= currPointedLedger.getEntries()) {
+ lastAckedPosition = new PositionImpl(nextPointedLedger.getLedgerId(), -1);
+ }
+ } else {
+ log.debug("No need to reset cursor: {}, current ledger is the last ledger.", cursor);
+ }
+ } else {
+ log.warn("Cursor: {} does not exist in the managed-ledger.", cursor);
+ }
+
+ if (!lastAckedPosition.equals((PositionImpl) cursor.getMarkDeletedPosition())) {
+ try {
+ log.info("Reset cursor:{} to {} since ledger consumed completely", cursor, lastAckedPosition);
+ if (lastConfirmedEntry.compareTo(lastAckedPosition) < 0) {
+ lastConfirmedEntry = lastAckedPosition;
Review comment:
It's better to fix the cursor update behavior, not the `lastConfirmedEntry`. We can treat the (x, -1) as a specific mark delete position for the cursor.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org