@Override
    public void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Stat stat,
            MetaStoreCallback<Void> callback) {
        if (log.isDebugEnabled()) {
            log.debug("[{}] [{}] Updating cursor info ledgerId={} mark-delete={}:{} lastActive={}",
                    ledgerName, cursorName, info.getCursorsLedgerId(), info.getMarkDeleteLedgerId(),
                    info.getMarkDeleteEntryId(), info.getLastActive());
        }

        String path = PREFIX + ledgerName + "/" + cursorName;
        byte[] content = compressCursorInfo(info);

        long expectedVersion;

        if (stat != null) {
            expectedVersion = stat.getVersion();
            if (log.isDebugEnabled()) {
                log.debug("[{}] Creating consumer {} on meta-data store with {}", ledgerName, cursorName, info);
            }
        } else {
            expectedVersion = -1;
            if (log.isDebugEnabled()) {
                log.debug("[{}] Updating consumer {} on meta-data store with {}", ledgerName, cursorName, info);
            }
        }
        store.put(path, content, Optional.of(expectedVersion))
                .thenAcceptAsync(optStat -> callback.operationComplete(null, optStat), executor
                        .chooseThread(ledgerName))
                .exceptionally(ex -> {
                    executor.executeOrdered(ledgerName,
                            () -> callback.operationFailed(getException(ex)));
                    return null;
                });
    }
@Test(timeOut = 20000)
    void updatingCursorNode() throws Exception {
        MetaStore store = new MetaStoreImpl(metadataStore, executor);

        metadataStore.put("/managed-ledgers/my_test", "".getBytes(), Optional.empty()).join();

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

        ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(1).build();
        store.asyncUpdateCursorInfo("my_test", "c1", info, null, new MetaStoreCallback<Void>() {
            public void operationFailed(MetaStoreException e) {
                promise.completeExceptionally(e);
            }

            public void operationComplete(Void result, Stat version) {
                // Update again using the version
                metadataStore.failConditional(new MetadataStoreException("error"), (op, path) ->
                        op == FaultInjectionMetadataStore.OperationType.PUT
                                && path.contains("my_test") && path.contains("c1")
                );

                ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(2).build();
                store.asyncUpdateCursorInfo("my_test", "c1", info, version, new MetaStoreCallback<Void>() {
                    public void operationFailed(MetaStoreException e) {
                        // ok
                        promise.complete(null);
                    }

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