diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index bcfb8585b8c12..01b3a9cb85280 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -166,6 +166,7 @@ enum State {
         NoLedger, // There is no metadata ledger open for writing
         Open, // Metadata ledger is ready
         SwitchingLedger, // The metadata ledger is being switched
+        Closing, // The managed cursor is closing
         Closed // The managed cursor has been closed
     }
 
@@ -439,7 +440,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
     public void asyncReadEntries(final int numberOfEntriesToRead, final ReadEntriesCallback callback,
             final Object ctx) {
         checkArgument(numberOfEntriesToRead > 0);
-        if (STATE_UPDATER.get(this) == State.Closed) {
+        if (isClosed()) {
             callback.readEntriesFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
             return;
         }
@@ -489,7 +490,7 @@ public void readEntryComplete(Entry entry, Object ctx) {
     public void asyncGetNthEntry(int n, IndividualDeletedEntries deletedEntries, ReadEntryCallback callback,
             Object ctx) {
         checkArgument(n > 0);
-        if (STATE_UPDATER.get(this) == State.Closed) {
+        if (isClosed()) {
             callback.readEntryFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
             return;
         }
@@ -554,7 +555,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
     @Override
     public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallback callback, Object ctx) {
         checkArgument(numberOfEntriesToRead > 0);
-        if (STATE_UPDATER.get(this) == State.Closed) {
+        if (isClosed()) {
             callback.readEntriesFailed(new CursorAlreadyClosedException("Cursor was already closed"), ctx);
             return;
         }
@@ -628,6 +629,10 @@ public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallbac
         }
     }
 
+    private boolean isClosed() {
+        return state == State.Closed || state == State.Closing;
+    }
+
     @Override
     public boolean cancelPendingReadRequest() {
         if (log.isDebugEnabled()) {
@@ -1351,7 +1356,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
         checkNotNull(position);
         checkArgument(position instanceof PositionImpl);
 
-        if (STATE_UPDATER.get(this) == State.Closed) {
+        if (isClosed()) {
             callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
             return;
         }
@@ -1567,7 +1572,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
 
     @Override
     public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallback callback, Object ctx) {
-        if (state == State.Closed) {
+        if (isClosed()) {
             callback.deleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
             return;
         }
@@ -1885,6 +1890,14 @@ private boolean shouldPersistUnackRangesToLedger() {
 
     private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
             MetaStoreCallback<Void> callback, boolean persistIndividualDeletedMessageRanges) {
+        if (state == State.Closed) {
+            ledger.getExecutor().execute(safeRun(() -> {
+                callback.operationFailed(new MetaStoreException(
+                        new ManagedLedgerException.CursorAlreadyClosedException(name + " cursor already closed")));
+            }));
+            return;
+        }
+
         // When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger,
         // hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed.
         ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() //
@@ -1919,13 +1932,14 @@ public void operationFailed(MetaStoreException e) {
 
     @Override
     public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object ctx) {
-        State oldState = STATE_UPDATER.getAndSet(this, State.Closed);
-        if (oldState == State.Closed) {
+        State oldState = STATE_UPDATER.getAndSet(this, State.Closing);
+        if (oldState == State.Closed || oldState == State.Closing) {
             log.info("[{}] [{}] State is already closed", ledger.getName(), name);
             callback.closeComplete(ctx);
             return;
         }
         persistPosition(-1, lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, callback, ctx);
+        STATE_UPDATER.set(this, State.Closed);
     }
 
     /**
@@ -2175,7 +2189,7 @@ boolean shouldCloseLedger(LedgerHandle lh) {
         long now = clock.millis();
         if ((lh.getLastAddConfirmed() >= config.getMetadataMaxEntriesPerLedger()
                 || lastLedgerSwitchTimestamp < (now - config.getLedgerRolloverTimeout() * 1000))
-                && STATE_UPDATER.get(this) != State.Closed) {
+                && (STATE_UPDATER.get(this) != State.Closed && STATE_UPDATER.get(this) != State.Closing)) {
             // It's safe to modify the timestamp since this method will be only called from a callback, implying that
             // calls will be serialized on one single thread
             lastLedgerSwitchTimestamp = now;