[fix][broker]Can't consume messages for a long time due to Entry Filter by poorbarcode · Pull Request #17782 · apache/pulsar

poorbarcode · 2022-09-21T22:40:48Z

Fixes:

Flaky-test: FilterEntryTest.testEntryFilterRescheduleMessageDependingOnConsumerSharedSubscription #16978

Motivation

When there are two consumers, users can specify the consumption behavior of each consumer by Entry filter:

case: consumer_1 can consume 60% of the messages, consumer_2 can consume 60% of the messages, and there is 10% intersection between consumer_1 and consumer_2

If returns FilterResult.RESCHEDULE for more than 10% of messages, then it's possible: some message that can only be consumed by consumer_1 keeps redelivered to consumer_2, and some message that can only be consumed by consumer_2 keeps redelivered to consumer_1. Then the problem occurs:

These messages can not be consumed anymore for a long time
The number of redeliveries of these messages has been increasing ( redelivery by Entry Filter ), see code below(line: 141):

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

Lines 140 to 148 in 8441f67

    
           } else if (filterResult == EntryFilter.FilterResult.RESCHEDULE) { 
        
               entriesToRedeliver.add((PositionImpl) entry.getPosition()); 
        
               entries.set(i, null); 
        
               // FilterResult will be always `ACCEPTED` when there is No Filter 
        
               // dont need to judge whether `hasFilter` is true or not. 
        
               this.filterRescheduledMsgs.add(entryMsgCnt); 
        
               entry.release(); 
        
               continue; 
        
           }

You can reproduce the problem by run FilterEntryTest.testEntryFilterRescheduleMessageDependingOnConsumerSharedSubscription 20 times

Modifications

When a message is redelivered by the same consumer more than 3 times, make that consumer pause to receive this message for 1 second.
Since tracking the consumption of all the messages cost memory too much, we trace only the earliest message.

Documentation

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

Matching PR in forked repository

PR in forked repository:

[ci-run][fix][broker]Can't consume messages for a long time due to Entry Filter poorbarcode/pulsar#8

github-actions · 2022-09-21T22:42:35Z

@poorbarcode Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

poorbarcode · 2022-09-21T23:35:51Z

This PR should merge into the following branches:

master
brnach-2.11

eolivelli · 2022-09-30T07:48:10Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+        if (!redeliveryTracker.hasRedeliveredEntry(entries)){
+            return super.getNextConsumer();
+        }
+        if (redeliveryTracker instanceof InMemoryAndPreventCycleFilterRedeliveryTracker tracker){


we should not use "instanceof"
we should leverage polimorphism

this comment seems unresolved

Already fixed

eolivelli · 2022-09-30T07:48:36Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+        }
+        if (redeliveryTracker instanceof InMemoryAndPreventCycleFilterRedeliveryTracker tracker){
+            if (tracker.pausedConsumerCount() == consumerSet.size()){
+                log.warn("No consumers are currently able to consume the first redelivery entry {}",


I think that this will be logged may times, and actually it is not a problem.

I am testing with some heavy load, in order to tell more

already fixed.

eolivelli · 2022-09-30T07:49:14Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+                return super.getNextConsumer();
+            } else {
+                Consumer nextConsumer = null;
+                while (true){


this looks like a endless loop in some cases.
we cannot keep this thread busy forever

already fixed.

...in/java/org/apache/pulsar/broker/service/InMemoryAndPreventCycleFilterRedeliveryTracker.java

eolivelli · 2022-09-30T09:00:31Z

...in/java/org/apache/pulsar/broker/service/InMemoryAndPreventCycleFilterRedeliveryTracker.java

+        pausedConsumers.clear();
+    }
+
+    private static int comparePosition(Position pos1, Position pos2) {


this method should go somewhere in Position or in some utility class
like comparePositionByEntryId

please check in the code if we already have something like that

Already fixed. I didn't find the right tool method, so I added a new one.

it seems you are not using the new utility method you introduce here

Already fixed.

this method is not used, we should drop it

already removed

eolivelli · 2022-09-30T09:02:18Z

...in/java/org/apache/pulsar/broker/service/InMemoryAndPreventCycleFilterRedeliveryTracker.java

+                return false;
+            }
+            // Automatically becomes invalid after 1s, because users may use time to filter the Entry.
+            return System.currentTimeMillis() - pauseTime < 1000;


this should be configurable, maybe next to ServiceConfiguration#dispatcherEntryFilterRescheduledMessageDelay ?

Already fixed. Could you please help me check whether DOC is suitable?

eolivelli · 2022-09-30T09:05:32Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RedeliveryTracker.java


    void clear();
+
+    default boolean hasRedeliveredEntry(List<Entry> entries) {


maybe the default implementation should be no-op, this way we don't add load to users who don't need this feature

I feel that since this method exists, it should provide the correct logic, so it should not be no-op

@codelipenghui what's your take on this ?

This method already moved into InMemoryAndPreventCycleFilterRedeliveryTracker, see #17782 (comment)

…ause time

poorbarcode changed the title ~~[ci-run][fix][broker]Can't consume messages for a long time due to Entry Filter~~ [fix][broker]Can't consume messages for a long time due to Entry Filter Sep 21, 2022

github-actions bot added the doc-label-missing label Sep 21, 2022

poorbarcode mentioned this pull request Sep 21, 2022

[ci-run][fix][broker]Can't consume messages for a long time due to Entry Filter poorbarcode/pulsar#8

Closed

github-actions bot added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Sep 21, 2022

poorbarcode force-pushed the flaky/FilterEntryTest.java branch 3 times, most recently from 02242d3 to 22baf48 Compare September 21, 2022 23:26

codelipenghui assigned poorbarcode Sep 22, 2022

codelipenghui requested review from Technoboy-, eolivelli and mattisonchao September 22, 2022 00:00

codelipenghui added this to the 2.12.0 milestone Sep 22, 2022

codelipenghui added release/2.11.1 release/2.10.3 and removed release/2.10.3 labels Sep 22, 2022

poorbarcode mentioned this pull request Sep 22, 2022

[doc][broker]PIP-187 Make FilterResult#RESCHEDULE to deprecated #17783

Closed

4 tasks

poorbarcode force-pushed the flaky/FilterEntryTest.java branch 2 times, most recently from 2a82948 to 3e48f88 Compare September 22, 2022 03:44

poorbarcode mentioned this pull request Sep 28, 2022

[ci] Move one test to the flaky suite #17819

Closed

4 tasks

poorbarcode force-pushed the flaky/FilterEntryTest.java branch from 3e48f88 to 2bb0869 Compare September 30, 2022 08:16

eolivelli requested changes Sep 30, 2022

View reviewed changes

eolivelli mentioned this pull request Sep 30, 2022

Enhance competing Consumers on filtered subscription datastax/pulsar#140

Closed

poorbarcode added 3 commits October 4, 2022 15:34

[fix][broker]Can't consume messages for a long time due to Entry Filter

acb1a50

address comment: use for-loop instead while-true & configurable max p…

6b3ea65

…ause time

explicitly remove consumer from cache

8d3986c

poorbarcode force-pushed the flaky/FilterEntryTest.java branch from 2bb0869 to 8d3986c Compare October 4, 2022 07:35

Draw out the util method of compare position

1c03853

poorbarcode removed the release/2.11.1 label Apr 10, 2023

Technoboy- modified the milestones: 3.1.0, 3.2.0 Jul 31, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

lhotari modified the milestones: 4.0.0, 4.1.0 Oct 14, 2024

lhotari added the release/4.0.1 label Oct 14, 2024

lhotari added release/4.0.2 and removed release/4.0.1 labels Dec 6, 2024

lhotari added release/4.0.3 and removed release/4.0.2 labels Jan 6, 2025

lhotari added release/4.0.4 and removed release/4.0.3 labels Feb 19, 2025

lhotari added release/4.0.5 and removed release/4.0.4 labels Mar 31, 2025

lhotari added release/4.0.6 and removed release/4.0.5 labels May 21, 2025

lhotari added release/4.0.7 and removed release/4.0.6 labels Jul 30, 2025

coderzc modified the milestones: 4.1.0, 4.2.0 Sep 1, 2025

lhotari added release/4.0.8 and removed release/4.0.7 labels Sep 27, 2025

lhotari added release/4.0.9 and removed release/4.0.8 labels Nov 11, 2025

lhotari added release/4.0.10 and removed release/4.0.9 labels Feb 9, 2026

	} else if (filterResult == EntryFilter.FilterResult.RESCHEDULE) {
	entriesToRedeliver.add((PositionImpl) entry.getPosition());
	entries.set(i, null);
	// FilterResult will be always `ACCEPTED` when there is No Filter
	// dont need to judge whether `hasFilter` is true or not.
	this.filterRescheduledMsgs.add(entryMsgCnt);
	entry.release();
	continue;
	}


		void clear();

		default boolean hasRedeliveredEntry(List<Entry> entries) {

Conversation

poorbarcode commented Sep 21, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Documentation

Matching PR in forked repository

Uh oh!

github-actions bot commented Sep 21, 2022

Uh oh!

poorbarcode commented Sep 21, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode Oct 4, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

poorbarcode commented Sep 21, 2022 •

edited

Loading

poorbarcode Oct 4, 2022 •

edited

Loading