Add unack-message threshold to restrict consumer for receiving messages without acknowledging-msg up to the threshold #48

rdhabalia · 2016-09-29T23:02:19Z

Motivation

If consumer keep tries to receive messages without acknowledging to broker then broker won't know last-deleted/received message entry. and if broker/consumer crashes/restarts in between then next time broker needs to send all the messages again as client didn't ack those messages.

Modifications

introduced configurable threshold maxUnackedMessagesPerConsumer limit till which consumer can consume messages without sending ack
once consumer reaches this maxUnackedMessagesPerConsumer limit, broker stops sending further messages until consumer sends acknowledgement for maxUnackedMessagesPerConsumer/2 messages.

Result

broker can block consumer if consumer doesn't send ack for maxUnackedMessagesPerConsumer number of messages.

yahoocla · 2016-09-29T23:02:21Z

CLA is valid!

yahoocla · 2016-09-29T23:02:21Z

CLA is valid!

yahoocla · 2016-09-29T23:02:21Z

CLA is valid!

sboobna · 2016-10-03T17:39:52Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/Consumer.java

-                    oldPermits);
+        if (!blockedConsumerOnUnackedMsgs) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Added more flow control message permits {} (old was: {})", this,


Since we are updating the messagePermits regardless, we should keep this log outside the if condition.

sboobna · 2016-10-03T17:40:19Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/Consumer.java

+                        additionalNumberOfMessages, oldPermits);
+            }
+            subscription.consumerFlow(this, additionalNumberOfMessages);
+            unackedMessages.addAndGet(additionalNumberOfMessages);


Shouldn't we update the unackedMessages while dispatching messages?

as we talked: we are not moving increment of unackedMessages at sendMessages(..) because this method is being called asynchronously and before it updates unackedMessages counter: if flowPermits(...) get triggered then it will add more permits and call dispatcher to dispatch more messages.

sboobna · 2016-10-03T17:46:07Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/Consumer.java

@@ -66,14 +66,19 @@
    private final ConcurrentOpenHashSet<PositionImpl> pendingAcks;

    private final ConsumerStats stats;
+
+    private final int maxUnackedMessages;


We should be able to disable this feature easily. What happens if we set maxUnackedMessages to 0.

To disable feature we can configure maxUnackedMessages =Integer.MAX_VALUE. Right now, configured value of maxUnackedMessages is 50000. Should we increase it to higher number.

It's better to use 0 for "unlimited"

Alright. I have added method shouldBlockConsumerOnUnackMsgs() which will enable this feature on sharedSubscription if maxUnackedMessages is > 0.

sboobna · 2016-10-04T01:22:50Z

conf/broker.conf

@@ -97,6 +97,10 @@ tlsTrustCertsFilePath=
 # Accept untrusted TLS certificate from client
 tlsAllowInsecureConnection=false

+# Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending
+# messages to consumer once, this limit reaches until consumer starts acknowledging messages back


You can mention here that setting the value to 0 will make it unlimited

yes.. added the documentation..

rdhabalia · 2016-10-05T20:32:28Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/Consumer.java

@@ -66,14 +66,19 @@
    private final ConcurrentOpenHashSet<PositionImpl> pendingAcks;


added change: to consider number of messages in a batch-msg while blocking consumer.

…es without acknowledging-msg up to the threshold

merlimat · 2016-10-08T00:24:10Z

@sboobna can you give a final pass?

sboobna

LGTM

Change initial version to 0.0.1

Signed-off-by: xiaolong.ran <ranxiaolong716@gmail.com> Fix the map of pendingReqs concurrent issue.

In pulsar cpp client, the default event time is set as minus value and caused error when use kafka consumer to consume message produced by pulsar cpp/cgo client. In this change, if we checked the event time is not set, using publish time to set it.

rdhabalia added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Sep 29, 2016

rdhabalia added this to the 1.15 milestone Sep 29, 2016

rdhabalia self-assigned this Sep 29, 2016

sboobna reviewed Oct 3, 2016

View reviewed changes

rdhabalia force-pushed the block_con branch 4 times, most recently from 58ca6d8 to 923636e Compare October 4, 2016 00:58

sboobna reviewed Oct 4, 2016

View reviewed changes

rdhabalia force-pushed the block_con branch 8 times, most recently from 0ec780f to f7c99ff Compare October 5, 2016 06:57

rdhabalia commented Oct 5, 2016

View reviewed changes

rdhabalia force-pushed the block_con branch 3 times, most recently from 4cfa842 to 910fcf5 Compare October 6, 2016 20:45

Add unack-message threshold to restrict consumer for receiving messag…

82bf1fc

…es without acknowledging-msg up to the threshold

rdhabalia force-pushed the block_con branch from 910fcf5 to 82bf1fc Compare October 7, 2016 19:23

merlimat approved these changes Oct 7, 2016

View reviewed changes

sboobna approved these changes Oct 8, 2016

View reviewed changes

merlimat merged commit 95b8239 into apache:master Oct 11, 2016

rdhabalia deleted the block_con branch November 11, 2016 23:04

sijie added a commit to sijie/pulsar that referenced this pull request Mar 4, 2018

Fix license header (apache#48)

103d20d

massakam pushed a commit to massakam/pulsar that referenced this pull request Aug 6, 2020

Merge pull request apache#48 from massakam/change-initial-version

8af5763

Change initial version to 0.0.1

hrsakai pushed a commit to hrsakai/pulsar that referenced this pull request Dec 10, 2020

Fix the map of pendingReqs concurrent issue (apache#48)

a7c1a04

Signed-off-by: xiaolong.ran <ranxiaolong716@gmail.com> Fix the map of pendingReqs concurrent issue.

xiaotongwang1 mentioned this pull request Aug 4, 2021

Pulsar 2.7.0+ KOP 2.7.2.x getPartitionedTopicMetadata timeout #11532

Closed

ZHr-UChiHa mentioned this pull request Jun 3, 2023

[Bug] python pulsar client create producer thread suspend because of deadlock apache/pulsar-client-python#129

Open

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add unack-message threshold to restrict consumer for receiving messages without acknowledging-msg up to the threshold #48

Add unack-message threshold to restrict consumer for receiving messages without acknowledging-msg up to the threshold #48

rdhabalia commented Sep 29, 2016

yahoocla commented Sep 29, 2016

yahoocla commented Sep 29, 2016

yahoocla commented Sep 29, 2016

sboobna Oct 3, 2016

sboobna Oct 3, 2016

rdhabalia Oct 3, 2016

sboobna Oct 3, 2016

rdhabalia Oct 3, 2016

merlimat Oct 3, 2016

rdhabalia Oct 4, 2016

sboobna Oct 4, 2016

rdhabalia Oct 5, 2016

rdhabalia Oct 5, 2016 •

edited

merlimat commented Oct 8, 2016

sboobna left a comment

		@@ -66,14 +66,19 @@
		private final ConcurrentOpenHashSet<PositionImpl> pendingAcks;

Add unack-message threshold to restrict consumer for receiving messages without acknowledging-msg up to the threshold #48

Add unack-message threshold to restrict consumer for receiving messages without acknowledging-msg up to the threshold #48

Conversation

rdhabalia commented Sep 29, 2016

Motivation

Modifications

Result

yahoocla commented Sep 29, 2016

yahoocla commented Sep 29, 2016

yahoocla commented Sep 29, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia Oct 5, 2016 • edited

Choose a reason for hiding this comment

merlimat commented Oct 8, 2016

sboobna left a comment

Choose a reason for hiding this comment

rdhabalia Oct 5, 2016 •

edited