Pull Consumer Updates by derekcollison · Pull Request #2776 · nats-io/nats-server

derekcollison · 2022-01-12T00:36:08Z

This change cleans up alot of the inner workings of pull consumers to be more resilient, better at waiting request management, and the ability to determine and watch interest cross account.

We also introduce new consumer config items to limit next msg requests' batch size and maximum wait times as well as indicate inactive thresholds.

We also now allow pull consumers that are ephemeral (they will auto-cleanup) and can have ack-all, not just ack explicit.

Signed-off-by: Derek Collison derek@nats.io

/cc @nats-io/core

Signed-off-by: Derek Collison <derek@nats.io>

Cleaned up code, made more consistent, utilize loopAndGather. Allow pull consumers to have AckAll as well as AckExplicit. Signed-off-by: Derek Collison <derek@nats.io>

… size and expiration. Signed-off-by: Derek Collison <derek@nats.io>

Signed-off-by: Derek Collison <derek@nats.io>

matthiashanel

LGTM
Change looks ok. But not super familiar, and a second set of eyes would be good

wallyqs · 2022-01-12T04:18:57Z


 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
+		o.expireWaiting()


Does this mean that doing sub.ConsumerInfo() will expire the pending requests?

Yes it will reclaim old requests here, and when we move to a wait state in the loop and gather, and we check them when we pop to deliver a message.

Not sure I understand how this all work, but when the user will try to subscribe, the library does a ConsumerInfo() (assuming a consumer name is provided by the user). So expiring existing pending requests just because an app wants to consume seem strange, but again, I may not understand well how pull consumers work in the server...

I think this is good approach, it was needed some way to trigger the expiration of accumulated pending requests that have just timed out without a client issuing a 'get next' request. Now when the state is confirmed by using nats consumer info for example, the expired requests would no longer be accounted in the NumWaiting field.

We used to expire pending requests sporadically and not on a regular time interval. I did not opt for a time interval per se but added a bunch of spots to where we do collect now.

Namely when we pop a request we now check inline, so no delivering to a request that we invalidate.

When the consumer's loopAndGather goes into a waiting state we check the pending list then as well.

When someone asks for consumer info on a pull consumer.

ripienaar · 2022-01-12T08:26:25Z

+	// If not durable determine the inactive threshold.
+	if !o.isDurable() {
+		if o.cfg.InactiveThreshold != 0 {
+			o.dthresh = o.cfg.InactiveThreshold


Would it make sense to constrain minimum inactive threshold around max ack time? They should expire some time after the last ack is expected, important for WQs etc

If this ack can be extended, that activity should also extend the inactivity timer(?)

Pull consumer logic is different, if we have seen a request in that window we are good, if not attempt to expire all invalid requests and if empty, then we consider inactive.

But if there are outstanding messages waiting on acks the consumer is essentially still valid - even with no interest or waiting pulls recently, it still has to wait for acks until the past the latest possible ack.

We do not follow that philosophy with ephemerals today. E.g. push consumers if the delivery subject loses interest for some period of time (5s default IIRC) we reclaim the consumer, regardless of pending acks.

This stays consistent but we could consider adding this, but I feel pretty strong we would not want to do that.

One app sends ack that gets lost and then disconnects, we will never clean up an ephemeral pull consumer if the logic requires the acks.

Not “never” just up to max ack waiting time.

Possibly but that would be inconsistent with how we do that today for push. So prefer to keep them the same for now and add in that logic if needed to a later date.

Also, if I am thinking about this correctly, if we only wait an additional max wait but take no other action, this feels like just reclaiming the consumer since we don't wait for non acked messages. IMO..

Yeah - my thinking is if you are processing messages from a WQ removing the consumer too early while there are still messages in flight could result in processed messages not being removed from tbe stream.

Agree it’s something we can park for a later time but seems important.

Good point but if its WQ and the consumer goes away that releases all of the messages anyway that were being held for that consumer.

kozlovic · 2022-01-12T17:51:58Z

 		}
-		// They are also required to be durable since otherwise we will not know when to
-		// clean them up.
-		if config.Durable == _EMPTY_ {


Just a note, this is making the nats.go client test suite fail because we were checking there that a durable name was required. Client tests will need some adjustment.

kozlovic · 2022-01-12T17:57:45Z


 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
+		o.expireWaiting()


Not sure I understand how this all work, but when the user will try to subscribe, the library does a ConsumerInfo() (assuming a consumer name is provided by the user). So expiring existing pending requests just because an app wants to consume seem strange, but again, I may not understand well how pull consumers work in the server...

Signed-off-by: Derek Collison <derek@nats.io>

…fnode. When we want to track service import response interest across a leafnode we need to send sub and unsub for all response _R_ subjects versus using a wildcard. Signed-off-by: Derek Collison <derek@nats.io>

Under load we could have a message committed to the underlying store when a consumer was being created and then it increase num pending again when the stream signals the consumers. This fix just remembers the last seq of the state when we calculate sgap and test before adding in the stream code. Signed-off-by: Derek Collison <derek@nats.io>

kozlovic

LGTM

Signed-off-by: Derek Collison <derek@nats.io>

derekcollison added 4 commits January 10, 2022 16:58

Update client

42818f0

Signed-off-by: Derek Collison <derek@nats.io>

Updated pull consumers.

5592d92

Cleaned up code, made more consistent, utilize loopAndGather. Allow pull consumers to have AckAll as well as AckExplicit. Signed-off-by: Derek Collison <derek@nats.io>

Add in ability to limit aspects of a pull request, specifically batch…

e12c8cd

… size and expiration. Signed-off-by: Derek Collison <derek@nats.io>

Add in ability to have ephemeral pull based consumers

279f31e

Signed-off-by: Derek Collison <derek@nats.io>

derekcollison requested review from kozlovic, matthiashanel and ripienaar January 12, 2022 00:36

matthiashanel approved these changes Jan 12, 2022

View reviewed changes

wallyqs reviewed Jan 12, 2022

View reviewed changes

ripienaar reviewed Jan 12, 2022

View reviewed changes

kozlovic reviewed Jan 12, 2022

View reviewed changes

Track interest properly across accounts for pull consumers

32c3c9e

Signed-off-by: Derek Collison <derek@nats.io>

derekcollison force-pushed the pull-consumer branch from 6796265 to 32c3c9e Compare January 12, 2022 20:17

Add in explicit subscription for import responses when bound to a lea…

43eff40

…fnode. When we want to track service import response interest across a leafnode we need to send sub and unsub for all response _R_ subjects versus using a wildcard. Signed-off-by: Derek Collison <derek@nats.io>

derekcollison force-pushed the pull-consumer branch from ddcc3b1 to 43eff40 Compare January 13, 2022 02:24

kozlovic approved these changes Jan 13, 2022

View reviewed changes

derekcollison added 2 commits January 13, 2022 09:14

Merge branch 'main' into pull-consumer

6619e7f

Merge from main

7dc99c3

Signed-off-by: Derek Collison <derek@nats.io>

derekcollison merged commit de5372e into main Jan 13, 2022

derekcollison deleted the pull-consumer branch January 13, 2022 18:26

aricart mentioned this pull request Jan 26, 2022

new pull consumer and ephemeral consumer options nats-io/nats-architecture-and-design#90

Open

15 tasks

kozlovic mentioned this pull request Aug 31, 2022

Allow AckAll for pull-based consumers nats-io/nats.go#1063

Merged

This was referenced Jan 19, 2023

Implement new pull consumer and ephemeral consumer options nats-io/nats-pure.rb#109

Open

Implement new pull consumer and ephemeral consumer options nats-io/nats.py#412

Closed

aradalvand mentioned this pull request Oct 2, 2025

fix: handle 408 Requests Pending responses for fetch requests nats-io/nats.net#973

Merged

Uh oh!

Uh oh!

Conversation

derekcollison commented Jan 12, 2022

Uh oh!

matthiashanel left a comment

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!

derekcollison Jan 12, 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!

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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kozlovic left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

derekcollison Jan 12, 2022 •

edited

Loading