[Broker] waitingCursors potential heap memory leak

[gaozhangmin]

Motivation

org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl#waitingCursors has memory leak problem, reproduced steps:

1. Create a consumer
2. Consumers is closed due to some error, without consuming any message.
3. Day by day, it will cause broker heap memory leak problem.

Reproduce test: I don't know why getWaitingCursorsCount aways 1 less than for loop count.

@Test
    public void testSub() throws Exception {
        final String topicName = "persistent://prop/ns-abc/stuckSubscriptionTopic";
        for (int i=0;i<10;i++) {
            Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
                    .subscriptionType(SubscriptionType.Failover).subscriptionName("test" + i).subscribe();
            consumer.close();
        }
        PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
        ManagedLedgerImpl ml = (ManagedLedgerImpl)(topicRef.ledger);
        assertEquals(9, ml.getWaitingCursorsCount());
    }

Detail:

1. When create consumer and connection opens.
   

   pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

   Lines 807 to 809 in a18f8a1

   	if (!(firstTimeConnect && hasParentConsumer) && conf.getReceiverQueueSize() != 0) {
   	increaseAvailablePermits(cnx, conf.getReceiverQueueSize());
   	}

2. Cursor was added to waitingCursors by org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#checkForNewEntries
   

   pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

   Lines 788 to 806 in a18f8a1

   	private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Object ctx) {
   	try {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] [{}] Re-trying the read at position {}", ledger.getName(), name, op.readPosition);
   	}
   	if (!hasMoreEntries()) {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] [{}] Still no entries available. Register for notification", ledger.getName(),
   	name);
   	}
   	// Let the managed ledger know we want to be notified whenever a new entry is published
   	ledger.waitingCursors.add(this);
   	} else {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] [{}] Skip notification registering since we do have entries available",
   	ledger.getName(), name);
   	}
   	}

3. But cursor were only removed when there were new messages. if consumer close without consuming messages, The cursor was not removed permantly.
   org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl#notifyCursors
   

   pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

   Lines 2219 to 2228 in a18f8a1

   	void notifyCursors() {
   	while (true) {
   	final ManagedCursorImpl waitingCursor = waitingCursors.poll();
   	if (waitingCursor == null) {
   	break;
   	}
   	executor.execute(safeRun(waitingCursor::notifyEntriesAvailable));
   	}
   	}

Modifications

1、remove cursor when consumer closed.

Documentation

Check the box below or label this PR directly (if you have committer privilege).

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)

[github-actions]

@gaozhangmin:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@gaozhangmin:Thanks for providing doc info!

[merlimat]

@gaozhangmin I'm not sure I see the case where the cursor is added multiple times:

ledger.getScheduledExecutor()
                        .schedule(() -> checkForNewEntries(op, callback, ctx),
                                config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);

This is called only the first time, as a kind of optimization to avoid inserting in the queue for cases where consumer and producers are very close. After the first time is unsuccessful, then we insert in the queue and wait for the notification. There shouldn't be more than 1 entry per cursor in that queue.

[gaozhangmin]

@gaozhangmin I'm not sure I see the case where the cursor is added multiple times:

ledger.getScheduledExecutor()
                        .schedule(() -> checkForNewEntries(op, callback, ctx),
                                config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);

This is called only the first time, as a kind of optimization to avoid inserting in the queue for cases where consumer and producers are very close. After the first time is unsuccessful, then we insert in the queue and wait for the notification. There shouldn't be more than 1 entry per cursor in that queue.

@merlimat You can verify by this test

@Test
    public void testSub() throws Exception {
        final String topicName = "persistent://prop/ns-abc/stuckSubscriptionTopic";
        for (int i=0;i<10000;i++) {
            Consumer<String> consumer1 = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
                    .subscriptionType(SubscriptionType.Failover).subscriptionName("test").subscribe();
            consumer1.close();
        }
        PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
        ManagedLedgerImpl ml = (ManagedLedgerImpl)(topicRef.ledger);
        System.out.println(ml.getWaitingCursorsCount());

    }

[liudezhi2098]

This maybe will not cause memory leak, it's not a periodic task. constantly creating and shutting down consumers can lead to this problem.

ledger.getScheduledExecutor()
                        .schedule(() -> checkForNewEntries(op, callback, ctx),
                                config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);

[gaozhangmin]

@codelipenghui @merlimat @lhotari @Jason918 @hangc0276 pTAL

[gaozhangmin]

/pulsarbot run-failure-checks

[hangc0276]

@liudezhi2098 Please help review this Pr, thanks a lot.

[gaozhangmin]

@merlimat PTAL

[gaozhangmin]

/pulsarbot run-failure-checks

[gaozhangmin]

/pulsarbot run-failure-checks