Motivation

Background

1. Pending read if no entries to read.

• Cursor sets itself into managedledger.waitingCursors if there are no entries to read.
• Managed ledger triggers cursor.notifyEntriesAvailable once new entries were added.

2, Double-check that it has more entries.

• The cursor does an additional check if there are no entries to read^[code-1].
• The cursor delays the second checking if the configuration newEntriesCheckDelayInMillis is larger than 0^[code-1].

3. The variable waitingReadOp of cursor

• It used to maintain an OpReadEntry to trigger a new read after getting notified that new entries were added.
• It prevents an issue from occurring, where more than one pending read is in progress

4. The variable waitingCursors of Managed Cursor

• It maintains the list of cursors that are waiting for new entries.

Issue 1: The same cursor was added to managedledger.waitingCursors thousands of times.

• Configurations
  • subscription type: Failover
• Reproduce flow
  • Add consumer(C1); it becomes an active consumer.
  • Add consumer(C2); C2 becomes the active consumer after a choice by the dispatcher
    • Triggers a scheduleReadOnActiveConsumer
    • (Highlight) The method scheduleReadOnActiveConsumer resets the variable cursor.waitingReadOp, but it forgets to remove the cursor from managedledger.waitingCursors
    • The issue occurs, which is the same as [Bug] Broker memory leak  #22157. You can reproduce the issue with the new test testWaitingCursorsCountAfterSwitchingActiveConsumers

Issue 2: Reused a recycled OpReadEntry, which causes repeated delivery and other unexpected issues

Issue 3: The method cursor.checkForNewEntries forgets to remove cursor from managedledger.waitingCursors.

• cursor.checkForNewEntries implements the double-check that checks whether there are entries to read.
• Clear cursor.waitingReadOp and trigger a reading if there are entries to read after the first check.
• The method forgot to remove the cursor from managedledger.waitingCursors after clearing the variable cursor.waitingReadOp

• [code-1]: https://github.com/apache/pulsar/blob/v4.0.5/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L1033-L1052

            if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
                op.recycle();
                callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
                return;
            }
            if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
                ledger.getScheduledExecutor()
                        .schedule(() -> checkForNewEntries(op, callback, ctx),
                                getConfig().getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
            } else {
                // If there's no delay, check directly from the same thread
                checkForNewEntries(op, callback, ctx);
            }

• [code-2]: https://github.com/apache/pulsar/blob/v4.0.5/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L1083-L1099

            if (hasMoreEntries()) {
                if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
                    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);
                } 
            }

Modifications

• Fix the 3 issues
• Instead of modifying managedledger.waitingCursors and cursor.waitingReadOp everywhere, Managed Cursor handles the variables itself

Documentation

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

Matching PR in forked repository

PR in forked repository: x