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 2020/06/30 22:04:49 UTC
[GitHub] [pulsar] merlimat commented on a change in pull request #7403: Avoid NPEs at ledger creation when DNS failures happen
merlimat commented on a change in pull request #7403:
URL: https://github.com/apache/pulsar/pull/7403#discussion_r448005570
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
##########
@@ -2283,73 +2283,78 @@ void internalFlushPendingMarkDeletes() {
void createNewMetadataLedger(final VoidCallback callback) {
ledger.mbean.startCursorLedgerCreateOp();
- ledger.asyncCreateLedger(bookkeeper, config, digestType, (rc, lh, ctx) -> {
- if (ledger.checkAndCompleteLedgerOpTask(rc, lh, ctx)) {
- return;
- }
+ try {
+ ledger.asyncCreateLedger(bookkeeper, config, digestType, (rc, lh, ctx) -> {
- ledger.getExecutor().execute(safeRun(() -> {
- ledger.mbean.endCursorLedgerCreateOp();
- 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)));
+ if (ledger.checkAndCompleteLedgerOpTask(rc, lh, ctx)) {
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);
- }
- });
+ ledger.getExecutor().execute(safeRun(() -> {
+ ledger.mbean.endCursorLedgerCreateOp();
+ 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)));
+ return;
}
- @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);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Created ledger {} for cursor {}", ledger.getName(), lh.getId(), name);
}
- });
- }));
- }, LedgerMetadataUtils.buildAdditionalMetadataForCursor(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);
+ }
+ });
+ }));
+ }, LedgerMetadataUtils.buildAdditionalMetadataForCursor(name));
+ } catch (Throwable t) {
+ log.error("[{}] Encountered unexpected error when creating cursor ledger", name, t);
+ callback.operationFailed(new ManagedLedgerException(t));
Review comment:
Correct. let me pull that back
----------------------------------------------------------------
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