Allow Consumer pausing and resuming consumption of individual partitions #417

JaapRood · 2019-07-05T10:46:11Z

To address #396.

…er than entire topic

JaapRood · 2019-07-05T10:55:45Z

I did a first draft of the required to see what the implications would be for implemention, public API, etc. In doing so I've identified some friction with the current public API of consumer.pause and consumer.resume.

Both already accept an array partitions to be specified with the topic name. The issue arises is what happens when omitting this partitions array, or including an empty one. The rest of the changes so far suggests that if the partitions would be empty, nothing would be paused, but so far the behaviour has been the opposite: the entire topic is paused.

The question here is whether we have to break compatibility to make this change or not, or whether we can find some reasonable way to maintain the current behaviour.

I'd propose the following:

when typeof partitions === undefined pause all current and future (can change through rebalance events) partitions
when Array.isArray(partitions) pause only those partitions, no partitions when empty
otherwise throw an error

I might give that approach a go and see if I can get the existing tests to pass, that should be a good indicator whether it's at all something we could consider.

… are defined

…er.run with the rest of the payload validation

JaapRood · 2019-07-05T14:20:47Z

The pause.spec.js tests are passing without any changes to the tests, so that must mean we're doing something right :)

However, the implementation also brought an issue to light with the tracking of only the paused partitions. When having paused "all" partitions through calling consumer.pause([{ topic }]), you then can't resume any partitions specifically. That's because the SubscriptionState does not have access to the partitions that have actually been assigned to the consumer group.

This could be solved by not only keeping a collection of paused topics, but also resumed. An alternative would be for the SubscriptionState to own the state describing what topics and partitions are assigned (currently in consumerGroup.topics and consumerGroup.memberAssignment), which given it's name suggests that might be the right thing to do anyway. That approach would also clean up consumerGroup.fetch which now has to join the paused with the assigned manually (although it's the only place where that needs to happen, so perhaps that's fine).

~~I'll have a look at that next, as well as what other pre-existing tests are failing. From there, we can start to add tests covering the new behaviour.~~

Edit: it makes more sense to get new behaviour tests in first, so we can actually prove the refactoring is necessary by showing failing tests!

tulios · 2019-07-08T08:37:48Z

Nice, and I agree with your proposal. I can try to get an early review today.

JaapRood · 2019-07-08T10:33:09Z

The version I just pushed fixes and tests the basic implementation of pausing and resuming single partitions.

It does not yet cover the case where we pause by only topic, to then resume a specific partition, which is currently unsupported. I'm not sure whether it's worth the refactoring required at this point to make that work, as I don't think it's proven this is an actual use case. I reckon you either choose topics as your unit or partitions in this case.

With the meat of the changes here, it's ready for review while I get some other tests passing, like the subscription state unit tests.

JaapRood · 2019-07-08T13:31:44Z

The PR should be in pretty good shape now, with fixed unit tests, a happy linter and documentation for the enhancement. It seems like the current failure of the tests is unrelated to the change and probably attributable to flakiness!

JaapRood · 2019-07-08T13:37:32Z

Just fyi, #421 will impact this, as it exposes the internals of SubscriptionState that have changed in this version. I'd recommend making these changes first, to then resolve how to expose it through the public API.

JaapRood · 2019-07-08T13:45:12Z

src/consumer/subscriptionState.js

+      } else if (Array.isArray(partitions) && !pausedForTopic.all) {
+        partitions.forEach(partition => pausedForTopic.partitions.delete(partition))
+      } else if (Array.isArray(partitions) && pausedForTopic.all) {
+        // TODO: consider whether we should actively track active topics, rather than paused ones, as to avoid this,


We'll probably want to either do something about this, or document it elsewhere. I don't think it's a proven use case yet, so documenting it and leaving the throw for now seems like the way to go.

I feel like we should probably make a note of this in an issue or something and bundle it into some future API breaking change instead, because the current behavior is pretty confusing (but totally the right call, given backwards compatibility).

Do you mean the API difference between calling with or withouts partitions specified? This comment was about the error we have to throw when we've previously paused all topics by not specifying partitions, to then try to resume a single partition. My idea so far has been that people use either full topic pausing, or per partition, but if this thrown Error is too confusing, we can go with bringing the current member assignment into SubscriptionState, which would allow mixing the two.

This could be solved by not only keeping a collection of paused topics, but also resumed. An alternative would be for the SubscriptionState to own the state describing what topics and partitions are assigned (currently in consumerGroup.topics and consumerGroup.memberAssignment), which given it's name suggests that might be the right thing to do anyway. That approach would also clean up consumerGroup.fetch which now has to join the paused with the assigned manually (although it's the only place where that needs to happen, so perhaps that's fine).

Maybe that's just the way to go anyway, as it would probably clean things up a bit, too. I'll have a look at how much is involved by having a small go at it.

Yeah, I was talking about the general problem of calling the public API with or without partitions. I'm not so sure that the assumption that people will either pause entire topics or individual partitions is valid, but I'm perfectly fine with us making the decision to go with it until we discover otherwise.

If you want to look into moving assigned partitions into SubscriptionState, you're very welcome to. I agree that it's likely the way to go, but I don't think it necessarily has to be done within the scope of this PR.

tulios · 2019-07-16T08:39:55Z

Just a note to myself, update the typescript types before releasing this.

JaapRood · 2019-07-16T09:53:29Z

@tulios how can I help with that in the future? Must say I haven't really used Typescript apart from some small experiments, but would like to offer complete PR's in the future!

tulios · 2019-07-18T08:20:15Z

@JaapRood we are also getting used with the typescript flow, so don't worry. I made this mistake several times by now. 😓

I am planning to add a PR template with a checklist of things we should be aware

Draft general changes required to pause and resume by partition, rath…

1cdf666

…er than entire topic

JaapRood added 4 commits July 5, 2019 14:42

Pause or resume all partitions of a topic when no specific partitions…

ea1d258

… are defined

Fix determination of which topics and partitions to fetch for

62aade9

Move the validation of partitions passed to consumer.pause and consum…

1528c56

…er.run with the rest of the payload validation

Pass the entire topic partitions payload to consumer group resume

4203d49

JaapRood added 2 commits July 8, 2019 12:16

Fix and verify pausing a single partition of a topic

a3a2475

Fix and verify resuming of a partition paused earlier

8f3c9ca

JaapRood marked this pull request as ready for review July 8, 2019 10:33

JaapRood added 3 commits July 8, 2019 14:51

Verify SubscriptionState is correctly implemented

21887c8

Document the behaviour of pausing / resuming individual partitions

9baa9e4

Remove unused code from subscription state test

a0bc3ac

JaapRood mentioned this pull request Jul 8, 2019

feat: add consumer.paused() to get the list of paused topics #421

Closed

JaapRood commented Jul 8, 2019

View reviewed changes

JaapRood mentioned this pull request Jul 8, 2019

Support for max.poll.records #420

Closed

Merge branch 'master' into feature/pause-partitions

9bfeff6

Nevon approved these changes Jul 12, 2019

View reviewed changes

JaapRood mentioned this pull request Jul 15, 2019

Move member assignment state into SubscriptionState and enable it to apply pause logic to concrete assignments #427

Closed

Reference tulios#427 in todo of SubscriptionState

f6e6a95

Nevon merged commit 2b3e159 into tulios:master Jul 15, 2019

JaapRood deleted the feature/pause-partitions branch July 15, 2019 09:03

JaapRood mentioned this pull request Jul 15, 2019

Consumer can not pause and resume individual partitions #396

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Allow Consumer pausing and resuming consumption of individual partitions #417

Allow Consumer pausing and resuming consumption of individual partitions #417

JaapRood commented Jul 5, 2019

JaapRood commented Jul 5, 2019 •

edited

JaapRood commented Jul 5, 2019 •

edited

tulios commented Jul 8, 2019

JaapRood commented Jul 8, 2019

JaapRood commented Jul 8, 2019

JaapRood commented Jul 8, 2019 •

edited

JaapRood Jul 8, 2019 •

edited

Nevon Jul 12, 2019

JaapRood Jul 12, 2019

Nevon Jul 12, 2019

tulios commented Jul 16, 2019

JaapRood commented Jul 16, 2019

tulios commented Jul 18, 2019

Allow Consumer pausing and resuming consumption of individual partitions #417

Allow Consumer pausing and resuming consumption of individual partitions #417

Conversation

JaapRood commented Jul 5, 2019

JaapRood commented Jul 5, 2019 • edited

JaapRood commented Jul 5, 2019 • edited

tulios commented Jul 8, 2019

JaapRood commented Jul 8, 2019

JaapRood commented Jul 8, 2019

JaapRood commented Jul 8, 2019 • edited

JaapRood Jul 8, 2019 • edited

Choose a reason for hiding this comment

Nevon Jul 12, 2019

Choose a reason for hiding this comment

JaapRood Jul 12, 2019

Choose a reason for hiding this comment

Nevon Jul 12, 2019

Choose a reason for hiding this comment

tulios commented Jul 16, 2019

JaapRood commented Jul 16, 2019

tulios commented Jul 18, 2019

JaapRood commented Jul 5, 2019 •

edited

JaapRood commented Jul 5, 2019 •

edited

JaapRood commented Jul 8, 2019 •

edited

JaapRood Jul 8, 2019 •

edited