[fix][broker] Fix consumer stops receiving messages when with large backlogs processing #22454

Technoboy- · 2024-04-07T12:54:39Z

Motivation

When the backlog size > managedLedgerCursorBackloggedThreshold (default 1000), the cursor will be set inactive first. so this will cause the non-durable cursor not to add to the waiting cursor. This leads to the consumer stopping to receive messages.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 3082 to 3090 in 2469b97

    
           private void checkBackloggedCursor(PersistentSubscription subscription) { 
        
               // activate caught up cursor which include consumers 
        
               if (!subscription.getConsumers().isEmpty() 
        
                       && subscription.getCursor().getNumberOfEntries() < backloggedCursorThresholdEntries) { 
        
                   subscription.getCursor().setActive(); 
        
               } else { 
        
                   subscription.getCursor().setInactive(); 
        
               } 
        
           }

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

Lines 3816 to 3824 in 2469b97

    
           public void addWaitingCursor(ManagedCursorImpl cursor) { 
        
               if (cursor instanceof NonDurableCursorImpl) { 
        
                   if (cursor.isActive()) { 
        
                       this.waitingCursors.add(cursor); 
        
                   } 
        
               } else { 
        
                   this.waitingCursors.add(cursor); 
        
               } 
        
           }

Documentation

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

…sing

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

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

lhotari · 2024-04-08T03:55:51Z

Great work on this fix @Technoboy- ! Just one minor comment in #22454 (comment) .

...ar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java

lhotari

LGTM. Good work @Technoboy

…acklogs processing (apache#22454)

…acklogs processing (#22454)

lhotari · 2024-04-08T13:50:09Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -308,7 +308,6 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor

        if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
            deactivateCursor();
-            topic.getManagedLedger().removeWaitingCursor(cursor);


just wondering about this removal now after the PR has been merged. It seems that whatever isResetCursor means that it would be skipped in that case. @Technoboy- is that a problem?

…acklogs processing (#22454) (cherry picked from commit 57a616e)

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

Fix consumer stops receiving messages when with large backlogs proces…

295b2a9

…sing

Technoboy- added the ready-to-test label Apr 7, 2024

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Apr 7, 2024

Fix consumer stops receiving messages when with large backlogs proces…

06c552d

…sing

dao-jun approved these changes Apr 7, 2024

View reviewed changes

lhotari requested changes Apr 8, 2024

View reviewed changes

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java Outdated Show resolved Hide resolved

lhotari reviewed Apr 8, 2024

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java Show resolved Hide resolved

lhotari mentioned this pull request Apr 8, 2024

[Bug] Consumer Stops Receiving Messages with Large Backlogs Post-Processing #22435

Closed

3 tasks

codelipenghui reviewed Apr 8, 2024

View reviewed changes

...ar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java Outdated Show resolved Hide resolved

...ar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java Outdated Show resolved Hide resolved

Technoboy- added 2 commits April 8, 2024 15:32

update

11f3094

address comment

4048169

Technoboy- self-assigned this Apr 8, 2024

Technoboy- added this to the 3.3.0 milestone Apr 8, 2024

Technoboy- added release/3.2.3 release/3.0.5 labels Apr 8, 2024

lhotari reviewed Apr 8, 2024

View reviewed changes

...ar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java Outdated Show resolved Hide resolved

lhotari approved these changes Apr 8, 2024

View reviewed changes

addressed comment

e501fa5

codelipenghui approved these changes Apr 8, 2024

View reviewed changes

Technoboy- merged commit 57a616e into apache:master Apr 8, 2024
49 of 50 checks passed

Technoboy- added a commit to Technoboy-/pulsar that referenced this pull request Apr 8, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

7259fd0

…acklogs processing (apache#22454)

Technoboy- added the cherry-picked/branch-3.2 label Apr 8, 2024

Technoboy- added a commit that referenced this pull request Apr 8, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

40329ee

…acklogs processing (#22454)

Technoboy- added the cherry-picked/branch-3.0 label Apr 8, 2024

lhotari reviewed Apr 8, 2024

View reviewed changes

lhotari mentioned this pull request Apr 9, 2024

[fix][broker] Check cursor state before adding it to the waitingCursors #22191

Merged

4 tasks

lhotari added the release/3.1.4 label Apr 9, 2024

lhotari pushed a commit that referenced this pull request Apr 9, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

4d8a1d8

…acklogs processing (#22454) (cherry picked from commit 57a616e)

lhotari added the cherry-picked/branch-3.1 label Apr 9, 2024

mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 15, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

fe4bc49

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 17, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

0a6a3f4

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 17, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

4be1e50

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

mukesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 19, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

de7f730

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

lhotari added release/2.11.5 release/2.10.7 labels Apr 22, 2024

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Apr 23, 2024

[fix][broker] Fix consumer stops receiving messages when with large b…

8f3e1fd

…acklogs processing (apache#22454) (cherry picked from commit 40329ee)

lhotari mentioned this pull request May 3, 2024

[Bug] After updating broker from 3.0.3 to 3.0.4, reader cannot receive new messages from compacted topics. #22634

Closed

3 tasks

lhotari mentioned this pull request May 21, 2024

[Bug] consumers stops receiving new messages due to invalid blockedConsumerOnUnackedMsgs state #22657

Open

3 tasks

TakaHiR07 mentioned this pull request Jul 11, 2024

[Bug][broker] Occur so many ERROR log in broker, which is confusing #23022

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][broker] Fix consumer stops receiving messages when with large backlogs processing #22454

[fix][broker] Fix consumer stops receiving messages when with large backlogs processing #22454

Technoboy- commented Apr 7, 2024

lhotari commented Apr 8, 2024

lhotari left a comment

lhotari Apr 8, 2024

	private void checkBackloggedCursor(PersistentSubscription subscription) {
	// activate caught up cursor which include consumers
	if (!subscription.getConsumers().isEmpty()
	&& subscription.getCursor().getNumberOfEntries() < backloggedCursorThresholdEntries) {
	subscription.getCursor().setActive();
	} else {
	subscription.getCursor().setInactive();
	}
	}

	public void addWaitingCursor(ManagedCursorImpl cursor) {
	if (cursor instanceof NonDurableCursorImpl) {
	if (cursor.isActive()) {
	this.waitingCursors.add(cursor);
	}
	} else {
	this.waitingCursors.add(cursor);
	}
	}

[fix][broker] Fix consumer stops receiving messages when with large backlogs processing #22454

[fix][broker] Fix consumer stops receiving messages when with large backlogs processing #22454

Conversation

Technoboy- commented Apr 7, 2024

Motivation

Documentation

lhotari commented Apr 8, 2024

lhotari left a comment

Choose a reason for hiding this comment

lhotari Apr 8, 2024

Choose a reason for hiding this comment