Prevent the consumption of messages for topics paused while fetch is in-flight #397
Conversation
| await waitFor(() => offsetsConsumed.length === messages.length, { delay: 50 }) | ||
| await sleep(50) | ||
|
|
||
| // Hope that we're now in an active fetch state? Something like FETCH_START might help |
There was a problem hiding this comment.
This looks like it's gonna be extremely flaky. We should indeed add a new instrumentation event.
There was a problem hiding this comment.
An instrumentation event would definitely be better, however in all past test runs since the first version of this was introduced (in #367) I haven't found a trace of failing once. That said, if Kafka responds a bunch slower through doing some compaction or whatever, the timing is most likely hosed.
There was a problem hiding this comment.
These things tend to become an issue in CI, even if they're reasonably reliable locally.
There was a problem hiding this comment.
I added FETCH_START, so far without a payload, as the only use case for it now doesn't require one to function. It's mirrored after GROUP_JOIN, with subject first, action second. Was a bit of a judgement call, as the BATCH ones do list the verb first. I reckoned subject first should help group them nicely when listing them, which seems as good a reason for something rather arbitrary as any other :)
| consumer.pause([{ topic: topicName }]) | ||
| await producer.send({ acks: 1, topic: topicName, messages }) // trigger completion of fetch | ||
|
|
||
| await sleep(200) |
There was a problem hiding this comment.
Rather than sleeping an arbitrary amount and hoping we've consumed the message by then, use the waitFor utility.
There was a problem hiding this comment.
The point is that no message has been consumed, that nothing has happened, so that makes a waitFor a bit trickier! Happy to hear if you've got any suggestions for that :)
There was a problem hiding this comment.
Ah, I should have read it more properly. Perhaps there's an event you could wait for having happened, such as END_BATCH_PROCESS to make sure that a fetch has happened without necessarily having to wait 200ms.
There was a problem hiding this comment.
Ended up going with FETCH, which seems to capture the use case correctly. Looked into END_BATCH_PROCESS, which worked for the unfixed implementation, but never triggered in the first as the filtered response was an empty batch, which get skipped!
|
Should have some time to look at making the test a bit more robust later today :) |
…er we're in a fetching state
|
@Nevon should be good for another look! |
|
Looks good to me! |
We ran into this bug, where if you pause a topic in the time between a requesting a fetch and receiving a response for it, the received messages are consumed, despite being the topic having been paused. A classic race condition, easily fixed by filtering for paused topics once again at response time.