@Override
    public void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, Map<String, String> properties,
            MetaStoreCallback<ManagedLedgerInfo> callback) {
        // Try to get the content or create an empty node
        String path = PREFIX + ledgerName;
        store.get(path)
                .thenAcceptAsync(optResult -> {
                    if (optResult.isPresent()) {
                        ManagedLedgerInfo info;
                        try {
                            info = parseManagedLedgerInfo(optResult.get().getValue());
                            info = updateMLInfoTimestamp(info);
                            callback.operationComplete(info, optResult.get().getStat());
                        } catch (InvalidProtocolBufferException e) {
                            callback.operationFailed(getException(e));
                        }
                    } else {
                        // Z-node doesn't exist
                        if (createIfMissing) {
                            log.info("Creating '{}'", path);

                            store.put(path, new byte[0], Optional.of(-1L))
                                    .thenAccept(stat -> {
                                        ManagedLedgerInfo.Builder ledgerBuilder = ManagedLedgerInfo.newBuilder();
                                        if (properties != null) {
                                            properties.forEach((k, v) -> {
                                                ledgerBuilder.addProperties(
                                                        MLDataFormats.KeyValue.newBuilder()
                                                                .setKey(k)
                                                                .setValue(v)
                                                                .build());
                                            });
                                        }
                                        callback.operationComplete(ledgerBuilder.build(), stat);
                                    }).exceptionally(ex -> {
                                        callback.operationFailed(getException(ex));
                                        return null;
                                    });
                        } else {
                            // Tried to open a managed ledger but it doesn't exist and we shouldn't creating it at this
                            // point
                            callback.operationFailed(new MetadataNotFoundException("Managed ledger not found"));
                        }
                    }
                }, executor.chooseThread(ledgerName))
                .exceptionally(ex -> {
                    try {
                        executor.executeOrdered(ledgerName,
                                () -> callback.operationFailed(getException(ex)));
                    } catch (RejectedExecutionException e) {
                        //executor maybe shutdown, use common pool to run callback.
                        CompletableFuture.runAsync(() -> callback.operationFailed(getException(ex)));
                    }
                    return null;
                });
    }
@Test(timeOut = 20000)
    void failInCreatingMLnode() throws Exception {
        MetaStore store = new MetaStoreImpl(metadataStore, executor);

        final CompletableFuture<Void> promise = new CompletableFuture<>();

        metadataStore.failConditional(new MetadataStoreException("error"), (op, path) ->
                op == FaultInjectionMetadataStore.OperationType.PUT
        );

        store.getManagedLedgerInfo("my_test", false, new MetaStoreCallback<MLDataFormats.ManagedLedgerInfo>() {
            public void operationFailed(MetaStoreException e) {
                promise.complete(null);
            }

            public void operationComplete(ManagedLedgerInfo result, Stat version) {
                promise.completeExceptionally(new Exception("Operation should have failed"));
            }
        });
        promise.get();
    }