Add consumer pause & resume #41
Conversation
|
👏 |
|
Looks good, I would just mention in the readme that the topic won't be fetched in the next cycle, you might get messages from the paused topic in the current batch. |
|
You can also add the signature of the methods in the readme and mention that the library currently doesn't support specific partitions. |
|
This works fine as long as you are subscribed to multiple topics, and don't pause all of them. If you do, however, it looks like the fetch loop becomes synchronous, so nothing else ever gets a chance to execute. |
|
@tulios: 🏖💻🍹 |
| this.subscriptionState.resume(topics) | ||
| } | ||
|
|
||
| paused() { |
There was a problem hiding this comment.
small stuff, but wouldn't a get paused() be a nicer API?
| @@ -0,0 +1,26 @@ | |||
| module.exports = class SubscriptionState { | |||
| constructor() { | |||
| this.pausedTopics = new Set() | |||
There was a problem hiding this comment.
Don't want any of those filthy duplicates in here. 💅
| @@ -26,7 +26,7 @@ module.exports = () => { | |||
| const availablePartitions = partitionMetadata.filter(p => p.leader >= 0) | |||
| const numAvailablePartitions = availablePartitions.length | |||
|
|
|||
| if (message.partition) { | |||
| if (message.partition !== null && message.partition !== undefined) { | |||
|
This is as fascinating as it is terrifying. I'm running my consumer and pausing/resuming by calling an HTTP endpoint. If I pause consuming of all topics, KafkaJS goes into a busy loop that won't let me call the endpoint to resume consuming. The culprit appears to be: kafkajs/src/consumer/consumerGroup.js Line 265 in b474e53 ConsumerGroup#fetch. requests will be an empty array, so my thinking was that maybe await Promise.all([]) resolves synchronously or something (doesn't make any sense, but nothing makes sense in this world). So I tried changing it to:
const results = await Promise.all(requests.length > 0 ? requests : [new Promise(resolve => process.nextTick(() => resolve([])))])This did not help. At this point I know that it resolves asynchronously, but still, express never got a chance to do anything. So, let's try with const results = await Promise.all(requests.length > 0 ? requests : [new Promise(resolve => setTimeout(() => resolve([]), 0))])This works. At this point someone needs to explain to me how the event loop works, because clearly it does not work the way I think it does. |
This avoids going into an infinite loop when there are no topics to fetch from.
|
I'm still confused, but at least I now know why Scheduling fetch using |
|
I've been running this for a while now, and it seems to be working fine. However, if you're running with a loglevel >= info, you'll produce an insane amount of logs, as each fetch loop logs, and without any topics to fetch from, it will basically produce logs as fast as it can. I think it makes sense to add some throttling to the fetch loop, at least when you're not subscribed to any topic. |
This also fixes a bug where if you specify
0as the partition for a message, the partitioner would fall back to using the hash of the key, as it would considermessage.partitionfalsy.