Re: [I] [Bug] Broker memory leak [pulsar]

["Technoboy- (via GitHub)" <gi...@apache.org>]

Technoboy- commented on issue #22157:
URL: https://github.com/apache/pulsar/issues/22157#issuecomment-1975986010

   There is a race condition between the `consumer.close` and `checkForNewEntries`.
   When the consumer is closed, the `waitingCursor` is empty.
   https://github.com/apache/pulsar/blob/ccc2ea67c192b4771c9af30c6eaf994fe01958e5/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java#L298-L313. 
   Then after executing line-311, the cursor was added to `waitingCursor`.
   
   ###  Reproduced test
   
   1. Add sleep time for `checkForNewEntries`, like 10s.  https://github.com/apache/pulsar/blob/ccc2ea67c192b4771c9af30c6eaf994fe01958e5/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L983-L989
   ![image](https://github.com/apache/pulsar/assets/6297296/6fdb69d7-96a0-4190-b8e6-bf8b4bd81f37)
   
   3. Then the below test will reproduce.
   ```
   @Test
       public void testWaitingCursors() throws Exception {
           final String ns = "prop/ns-test";
           admin.namespaces().createNamespace(ns, 2);
           final String topicName = "persistent://prop/ns-test/testWaitingCursors";
           admin.topics().createNonPartitionedTopic(topicName);
           final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
                   .subscriptionMode(SubscriptionMode.NonDurable)
                   .subscriptionType(SubscriptionType.Exclusive)
                   .subscriptionName("sub-2").subscribe();
           final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
           producer.send("test");
           producer.close();
           final String broker = admin.lookups().lookupTopic(topicName);
           final Optional<Topic> topic = pulsar.getBrokerService().getTopic(topicName, false).join();
           assertNotNull(topic.get());
           PersistentTopic persistentTopic = (PersistentTopic) topic.get();
           final PersistentSubscription subscription = persistentTopic.getSubscription("sub-2");
           NonDurableCursorImpl cursor = (NonDurableCursorImpl) subscription.getCursor();
           final Message<String> receive = consumer.receive();
           assertEquals("test", receive.getValue());
           consumer.close();
           while (true) {
               ManagedLedgerImpl ledger = (ManagedLedgerImpl) cursor.getManagedLedger();
               log.info("waitingCursorsCount : {}", ledger.getWaitingCursorsCount());
               Thread.sleep(5 * 1000);
           }
       }
   ```


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org