Fixes #24566

Motivation

The test ReplicatorTest.testCloseReplicatorStartProducer was consistently failing due to critical issues introduced in PR #24551. Investigation (shared in #24566 (comment) by @3pacccccc ) revealed that the changes led to infinite recursive calls in ManagedCursorImpl#cancelPendingReadRequest, causing broker instability and test failures.

The root cause was improper state management for pending read operations, particularly when:

• Multiple threads attempt to cancel pending read requests concurrently
• Cursors are closed while read operations are pending
• OpReadEntry instances are recycled and reused incorrectly
• Cursor state transitions occur without proper synchronization
• Cursor deletion and closure operations race with active read operations

Modifications

This PR addresses the state management issues through several key changes:

Core Synchronization Fixes

1. Reverted problematic synchronization approach: Removed the complex pendingReadOpMutex and DelayCheckForNewEntriesTask implementation added in [fix][broker] Fix Broker OOM due to too many waiting cursors and reuse a recycled OpReadEntry incorrectly #24551 that caused infinite recursion, returning to a more robust compare-and-swap based solution.

2. Added operation ID tracking: Introduced an id field to OpReadEntry instances to prevent recycled operations from being processed incorrectly. This ensures that only the correct operation instance is handled when checking for new entries.

3. Improved atomic state transitions: Enhanced the CAS (Compare-And-Swap) operations using getAndUpdate() with lambda functions that verify both the operation instance and its ID before making state changes:

                OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
                        current -> {
                            if (current == op && current.id == opReadId) {
                                // update the value to null to cancel the waiting read op
                                return null;
                            } else {
                                // keep the current waiting read op value
                                return current;
                            }
                        });
                // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
                // cleared from the waitingReadOp field and therefore "cancelled"
                if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
                    if (log.isDebugEnabled()) {
                        log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
                                name, op.readPosition);
                    }
                    PENDING_READ_OPS_UPDATER.incrementAndGet(this);
                    ledger.asyncReadEntries(op);

Enhanced State Management System

4. Comprehensive state model enhancement: Significantly expanded the cursor State enum with:

   • New states: Deleting, Deleted, DeletingFailed for better deletion lifecycle tracking
   • Added closedState boolean field and isClosed() method for more robust state checking
   • Added isDeletingOrDeleted() method for deletion state validation
   • Enhanced state constructor to properly categorize closed vs. open states

5. Thread-safe state transition methods: Implemented atomic state transition utilities:

   • changeStateIfNotClosed() - prevents state changes on closed cursors
   • changeStateIfNotDeletingOrDeleted() - prevents state changes on deleting/deleted cursors
   • Enhanced trySetStateToClosing() with atomic operations

Cursor Lifecycle Management

6. Proper cursor closure handling: Added comprehensive cursor closure logic:

   • closeWaitingCursor() - handles normal cursor closure scenarios
   • cancelWaitingCursorsWhenDeactivated() - handles cursor deactivation scenarios
   • internalCloseWaitingCursor() - unified internal implementation with exception supplier pattern
   • Added sentinel value (WAITING_READ_OP_FOR_CLOSED_CURSOR) to handle closed cursor scenarios

7. Enhanced cursor deactivation: Updated setInactive() to properly cancel waiting cursors when deactivated, preventing orphaned read operations.

8. Robust deletion lifecycle: Improved asyncDeleteCursorLedger() with:

   • Proper state transitions through deletion lifecycle
   • Enhanced retry logic with backoff for failed deletions
   • Better error handling and state recovery for deletion failures
   • Prevention of duplicate deletion attempts

Exception Handling Improvements

9. New exception types: Added CursorDeactivatedWaitCallbackException for better error propagation when cursors are deactivated without properly cancelling pending requests.

10. Enhanced error propagation: Improved error handling throughout the cursor lifecycle to ensure proper cleanup and user notification.

Managed Ledger Integration

11. Enhanced cursor deletion in ManagedLedgerImpl: Improved asyncDeleteCursor() with:

    • Proper state checking before deletion attempts
    • Separate handling for durable vs. non-durable cursors
    • Better error recovery and state management
    • Prevention of concurrent deletion attempts

12. Enhanced waiting cursor registration: Implemented thread-safe registration/deregistration of waiting cursors to prevent race conditions when adding or removing cursors from the waiting list, with improved debug logging.

Specialized Cursor Implementation Updates

13. Non-durable cursor improvements: Updated NonDurableCursorImpl.asyncClose() to properly close waiting cursors and set inactive state.

14. Read-only cursor improvements: Updated ReadOnlyCursorImpl.asyncClose() with the same proper cleanup logic.

Misc refactoring in ManagedCursorImpl

15. In the previous code, the state field was sometimes read directly and sometimes with STATE_UPDATER.get. Similarly, it was sometimes set directly and mostly with STATE_UPDATER.set. There's no need to use STATE_UPDATER.get and STATE_UPDATER.set. The volatile state field can be read and set directly without STATE_UPDATER.
    STATE_UPDATER is only needed for CAS operations on the state field.

Key Technical Improvements

• Eliminated infinite recursion: The previous implementation could enter infinite recursive calls in cancelPendingReadRequest(), which is now resolved
• Thread-safe operation lifecycle: OpReadEntry instances are properly tracked through their lifecycle to prevent use-after-recycle scenarios
• Comprehensive state management: New state model provides better visibility and control over cursor lifecycle
• Enhanced error handling: Cursor closure and deletion events now properly propagate appropriate exceptions to pending operations
• Performance optimization: Reduced lock contention by using atomic operations instead of synchronized blocks where possible
• Improved debugging: Added comprehensive debug logging for cursor registration/deregistration operations
• Robust deletion handling: Deletion operations now properly handle failures, retries, and state recovery

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, specifically:

• ReplicatorTest.testCloseReplicatorStartProducer - verifies proper cleanup when replicator cursors are closed
• FailoverSubscriptionTest.testWaitingCursorsCountAfterSwitchingActiveConsumers - verifies that each cursor doesn't get added more than once into ManagedLedgerImpl's waiting cursors list
• Existing managed ledger tests that exercise concurrent read operations
• Integration tests that stress-test cursor state management under load
• Tests covering cursor deletion and state transitions

The modifications also include timeout adjustments for the test to account for the improved error handling paths.

Impact Assessment

This change affects:

• Threading model: Improved thread safety in cursor state management with atomic state transitions
• Error handling: Better propagation of cursor closure and deactivation exceptions
• Performance: Reduced lock contention in high-throughput scenarios through atomic operations
• Lifecycle management: More robust cursor deletion and closure handling
• State consistency: Enhanced state model prevents invalid state transitions
• Resource cleanup: Better cleanup of waiting cursors and pending operations

The changes are backward compatible and do not affect public APIs or configuration defaults.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete