rdhabalia commented on code in PR #22842: URL: https://github.com/apache/pulsar/pull/22842#discussion_r1626782396
########## managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java: ########## @@ -378,63 +378,77 @@ public void asyncOpen(final String name, final ManagedLedgerConfig config, final ledgers.computeIfAbsent(name, (mlName) -> { // Create the managed ledger CompletableFuture<ManagedLedgerImpl> future = new CompletableFuture<>(); - BookKeeper bk = bookkeeperFactory.get( - new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(), - config.getBookKeeperEnsemblePlacementPolicyProperties())); - final ManagedLedgerImpl newledger = config.getShadowSource() == null - ? new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker) - : new ShadowManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, - mlOwnershipChecker); - PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger); - pendingInitializeLedgers.put(name, pendingLedger); - newledger.initialize(new ManagedLedgerInitializeLedgerCallback() { - @Override - public void initializeComplete() { - log.info("[{}] Successfully initialize managed ledger", name); - pendingInitializeLedgers.remove(name, pendingLedger); - future.complete(newledger); - - // May need to update the cursor position - newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger(); - // May need to trigger offloading - if (config.isTriggerOffloadOnTopicLoad()) { - newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE); - } + CompletableFuture<BookKeeper> bkFuture = createBookKeeperClient(config); + bkFuture.handle((bk, ex) -> { + if (ex != null) { + future.completeExceptionally(ex); + return null; } + final ManagedLedgerImpl newledger = config.getShadowSource() == null + ? new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker) + : new ShadowManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, + mlOwnershipChecker); + PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger); + pendingInitializeLedgers.put(name, pendingLedger); + newledger.initialize(new ManagedLedgerInitializeLedgerCallback() { + @Override + public void initializeComplete() { + log.info("[{}] Successfully initialize managed ledger", name); + pendingInitializeLedgers.remove(name, pendingLedger); + future.complete(newledger); - @Override - public void initializeFailed(ManagedLedgerException e) { - if (config.isCreateIfMissing()) { - log.error("[{}] Failed to initialize managed ledger: {}", name, e.getMessage()); + // May need to update the cursor position + newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger(); + // May need to trigger offloading + if (config.isTriggerOffloadOnTopicLoad()) { + newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE); + } } - // Clean the map if initialization fails - ledgers.remove(name, future); + @Override + public void initializeFailed(ManagedLedgerException e) { + if (config.isCreateIfMissing()) { + log.error("[{}] Failed to initialize managed ledger: {}", name, e.getMessage()); + } - if (pendingInitializeLedgers.remove(name, pendingLedger)) { - pendingLedger.ledger.asyncClose(new CloseCallback() { - @Override - public void closeComplete(Object ctx) { - // no-op - } + // Clean the map if initialization fails + ledgers.remove(name, future); - @Override - public void closeFailed(ManagedLedgerException exception, Object ctx) { - log.warn("[{}] Failed to a pending initialization managed ledger", name, exception); - } - }, null); - } + if (pendingInitializeLedgers.remove(name, pendingLedger)) { + pendingLedger.ledger.asyncClose(new CloseCallback() { + @Override + public void closeComplete(Object ctx) { + // no-op + } - future.completeExceptionally(e); - } - }, null); + @Override + public void closeFailed(ManagedLedgerException exception, Object ctx) { + log.warn("[{}] Failed to a pending initialization managed ledger", name, exception); + } + }, null); + } + future.completeExceptionally(e); + } + }, null); + return null; + }); return future; }).thenAccept(ml -> callback.openLedgerComplete(ml, ctx)).exceptionally(exception -> { callback.openLedgerFailed((ManagedLedgerException) exception.getCause(), ctx); return null; }); } + private CompletableFuture<BookKeeper> createBookKeeperClient(ManagedLedgerConfig config) { + CompletableFuture<BookKeeper> future = new CompletableFuture<>(); + scheduledExecutor.execute(() -> { Review Comment: > I think the culprit of the issue is in the BookkeeperFactoryForCustomEnsemblePlacementPolicy interface Yes. But that's something that depends on other system which is BookKeeper in this case, Upgrading bk version will not be possible for all Pulsar stable version. Let's merge this patch and I can make bookkeeper enhancement to fix this issue for future version. -- 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-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org