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 2022/05/05 13:26:36 UTC
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15425: [PIP 81] Part-1 Split createNewMetadataLedger into multiple methods for reuse
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org