codelipenghui commented on code in PR #15425:
URL: https://github.com/apache/pulsar/pull/15425#discussion_r865903000
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2786,6 +2787,9 @@ public void operationComplete(Void result, Stat stat) {
@Override
public void operationFailed(MetaStoreException e) {
log.warn("[{}] Failed to update consumer {}",
ledger.getName(), name, e);
+ // it means it failed to switch the newly created ledger so,
it should be
+ // deleted to prevent leak
+ deleteLedger(lh);
Review Comment:
Here is persistent data to metadata store failed, why need to delete the
ledger? We can only delete the ledger after persistent to metadata store
successfully?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2544,63 +2576,32 @@ void createNewMetadataLedger(final VoidCallback
callback) {
if (rc != BKException.Code.OK) {
log.warn("[{}] Error creating ledger for cursor {}: {}",
ledger.getName(), name,
BKException.getMessage(rc));
- callback.operationFailed(new
ManagedLedgerException(BKException.getMessage(rc)));
+ future.completeExceptionally(new
ManagedLedgerException(BKException.getMessage(rc)));
return;
}
if (log.isDebugEnabled()) {
log.debug("[{}] Created ledger {} for cursor {}",
ledger.getName(), lh.getId(), name);
}
- // Created the ledger, now write the last position
- // content
- MarkDeleteEntry mdEntry = lastMarkDeleteEntry;
- persistPositionToLedger(lh, mdEntry, new VoidCallback() {
- @Override
- public void operationComplete() {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Persisted position {} for cursor
{}", ledger.getName(),
- mdEntry.newPosition, name);
- }
- switchToNewLedger(lh, new VoidCallback() {
- @Override
- public void operationComplete() {
- callback.operationComplete();
- }
-
- @Override
- public void operationFailed(ManagedLedgerException
exception) {
- // it means it failed to switch the newly
created ledger so, it should be
- // deleted to prevent leak
- bookkeeper.asyncDeleteLedger(lh.getId(), (int
rc, Object ctx) -> {
- if (rc != BKException.Code.OK) {
- log.warn("[{}] Failed to delete orphan
ledger {}", ledger.getName(),
- lh.getId());
- }
- }, null);
- callback.operationFailed(exception);
- }
- });
- }
-
- @Override
- public void operationFailed(ManagedLedgerException
exception) {
- log.warn("[{}] Failed to persist position {} for
cursor {}", ledger.getName(),
- mdEntry.newPosition, name);
-
- ledger.mbean.startCursorLedgerDeleteOp();
- bookkeeper.asyncDeleteLedger(lh.getId(), new
DeleteCallback() {
- @Override
- public void deleteComplete(int rc, Object ctx) {
- ledger.mbean.endCursorLedgerDeleteOp();
- }
- }, null);
- callback.operationFailed(exception);
- }
- });
+ future.complete(lh);
}));
}, LedgerMetadataUtils.buildAdditionalMetadataForCursor(name));
+
+ return future;
}
+ private void deleteLedger(LedgerHandle ledgerHandle) {
Review Comment:
```suggestion
private void deleteLedgerAsync(LedgerHandle ledgerHandle) {
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2544,63 +2576,32 @@ void createNewMetadataLedger(final VoidCallback
callback) {
if (rc != BKException.Code.OK) {
log.warn("[{}] Error creating ledger for cursor {}: {}",
ledger.getName(), name,
BKException.getMessage(rc));
- callback.operationFailed(new
ManagedLedgerException(BKException.getMessage(rc)));
+ future.completeExceptionally(new
ManagedLedgerException(BKException.getMessage(rc)));
return;
}
if (log.isDebugEnabled()) {
log.debug("[{}] Created ledger {} for cursor {}",
ledger.getName(), lh.getId(), name);
}
- // Created the ledger, now write the last position
- // content
- MarkDeleteEntry mdEntry = lastMarkDeleteEntry;
- persistPositionToLedger(lh, mdEntry, new VoidCallback() {
- @Override
- public void operationComplete() {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Persisted position {} for cursor
{}", ledger.getName(),
- mdEntry.newPosition, name);
- }
- switchToNewLedger(lh, new VoidCallback() {
- @Override
- public void operationComplete() {
- callback.operationComplete();
- }
-
- @Override
- public void operationFailed(ManagedLedgerException
exception) {
- // it means it failed to switch the newly
created ledger so, it should be
- // deleted to prevent leak
- bookkeeper.asyncDeleteLedger(lh.getId(), (int
rc, Object ctx) -> {
- if (rc != BKException.Code.OK) {
- log.warn("[{}] Failed to delete orphan
ledger {}", ledger.getName(),
- lh.getId());
- }
- }, null);
- callback.operationFailed(exception);
- }
- });
- }
-
- @Override
- public void operationFailed(ManagedLedgerException
exception) {
- log.warn("[{}] Failed to persist position {} for
cursor {}", ledger.getName(),
- mdEntry.newPosition, name);
-
- ledger.mbean.startCursorLedgerDeleteOp();
- bookkeeper.asyncDeleteLedger(lh.getId(), new
DeleteCallback() {
- @Override
- public void deleteComplete(int rc, Object ctx) {
- ledger.mbean.endCursorLedgerDeleteOp();
- }
- }, null);
- callback.operationFailed(exception);
- }
- });
+ future.complete(lh);
}));
}, LedgerMetadataUtils.buildAdditionalMetadataForCursor(name));
+
+ return future;
}
+ private void deleteLedger(LedgerHandle ledgerHandle) {
Review Comment:
+1
--
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]
