-
-
Notifications
You must be signed in to change notification settings - Fork 525
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Prevent the consumption of messages for topics paused while fetch is in-flight #397
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The implementation looks good. It's just the test that could use some improvement.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This looks like it's gonna be extremely flaky. We should indeed add a new instrumentation event.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
These things tend to become an issue in CI, even if they're reasonably reliable locally.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.