mattisonchao commented on code in PR #15425:
URL: https://github.com/apache/pulsar/pull/15425#discussion_r865470173
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2532,10 +2532,42 @@ void internalFlushPendingMarkDeletes() {
void createNewMetadataLedger(final VoidCallback callback) {
ledger.mbean.startCursorLedgerCreateOp();
+ doCreateNewMetadataLedger().thenAccept(newLedgerHandle -> {
+ MarkDeleteEntry mdEntry = lastMarkDeleteEntry;
+ // Created the ledger, now write the last position content
+ persistPositionToLedger(newLedgerHandle, mdEntry, new
VoidCallback() {
+ @Override
+ public void operationComplete() {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Persisted position {} for cursor {}",
ledger.getName(),
+ mdEntry.newPosition, name);
+ }
+ switchToNewLedger(newLedgerHandle, callback);
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException exception) {
+ log.warn("[{}] Failed to persist position {} for cursor
{}", ledger.getName(),
+ mdEntry.newPosition, name);
+
+ deleteLedger(newLedgerHandle);
+ callback.operationFailed(exception);
+ }
+ });
+ }).whenComplete((result, e) -> {
+ ledger.mbean.endCursorLedgerCreateOp();
+ if (e != null) {
+ callback.operationFailed(createManagedLedgerException(e));
+ }
+ });
+ }
+ private CompletableFuture<LedgerHandle> doCreateNewMetadataLedger() {
+ CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
ledger.asyncCreateLedger(bookkeeper, config, digestType, (rc, lh, ctx)
-> {
if (ledger.checkAndCompleteLedgerOpTask(rc, lh, ctx)) {
+ future.complete(null);
Review Comment:
looks like there is a risk of NPE.
##########
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:
Should we return `CompletableFuture` to make this method more reusable?
--
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]
