fix: handle errors gracefully instead of panicking in async producer - #3599
fix: handle errors gracefully instead of panicking in async producer#3599willy182 wants to merge 3 commits into
Conversation
puellanivis
left a comment
puellanivis
left a comment
There was a problem hiding this comment.
We can have a little bit of copying… as a treat.
But if we copy some code, and then mutate it, we should also look into how that mutation means we might be able to simplify things.
| sendResponse := func(response *ProduceResponse, err error) { | ||
| pending <- &brokerProducerResponse{ | ||
| set: mutedSet, | ||
| err: err, | ||
| res: response, | ||
| } | ||
| wg.Done() | ||
| } | ||
| sendResponse(nil, err) |
There was a problem hiding this comment.
It makes no sense to pass response = nil as a parameter here, nor set it as a field in the &brokerProducerRespnse as nil is going to be the automatic zero value.
Additionally, there’s no reason to pass err as a parameter either, it’s already available by closure, and we’re not mutating it anyways.
| sendResponse := func(response *ProduceResponse, err error) { | ||
| pending <- &brokerProducerResponse{ | ||
| set: mutedSet, | ||
| err: err, | ||
| res: response, | ||
| } | ||
| wg.Done() | ||
| } | ||
| sendResponse(nil, err) |
There was a problem hiding this comment.
One can simply call an anonymous function without needing to pass it through a temporary named variable.
| sendResponse := func(response *ProduceResponse, err error) { | |
| pending <- &brokerProducerResponse{ | |
| set: mutedSet, | |
| err: err, | |
| res: response, | |
| } | |
| wg.Done() | |
| } | |
| sendResponse(nil, err) | |
| func(response *ProduceResponse, err error) { | |
| pending <- &brokerProducerResponse{ | |
| set: mutedSet, | |
| err: err, | |
| res: response, | |
| } | |
| wg.Done() | |
| }(nil, err) |
| // Count the in flight requests to know when we can close the pending channel safely | ||
| wg.Add(1) | ||
| // capture the muted set. unmuting is deferred to handleResponse. | ||
| mutedSet := set |
There was a problem hiding this comment.
We’re using go1.25 https://github.com/IBM/sarama/blob/main/go.mod#L3
Capturing a loop variable in an iteration variable if it might escape isn’t necessary.
| } | ||
|
|
||
| req := ps.buildRequest() | ||
| req, _ := ps.buildRequest() |
There was a problem hiding this comment.
Do not throw away errors in tests. This is the fastest way to have an issue that could have been caught in tests get missed until it’s released to a deployment.
I already changed, please review again |
|
I’m curious when and how you’ve hit this panic? |
puellanivis
left a comment
puellanivis
left a comment
There was a problem hiding this comment.
Other than this last lint, the code seems reasonable. The larger scope I leave to the team.
| wg.Add(1) | ||
| pending <- &brokerProducerResponse{ | ||
| set: set, | ||
| err: err, | ||
| } | ||
| wg.Done() |
There was a problem hiding this comment.
I spot now that the wg.Add and wg.Done are unnecessary. The WaitGroup is defined on line 984, so this wait group control is definitely per-function-call, and since the send on pending blocks execution of this goroutine, and so the for loop cannot exit while this is executing, there’s no need to use wait group coördination.
You can spot below that the wg.Done is executed at an indeterminate point in the future by the async producer, but we’re in a different situation in this block.
puellanivis
left a comment
puellanivis
left a comment
There was a problem hiding this comment.
Approving just the code. The logic and fit with design lies with the main contributors.
|
I’m still curious when and how you’ve hit this panic, because we should never fail to encode requests, or whether this is just a speculative PR because you spotted panic in the code? |
|
@dnwe Honestly, I haven't hit this panic in production. I spotted it while reviewing the codebase and realized that a panic here could crash the entire application unnecessarily. Since there's already a working error propagation path., it felt like a straightforward improvement to replace the panic with a proper error return - defensinve programming rather than a bug fix. But if I guess, maybe this Panic case could happen because the encoding/compression process failed. |
|
OK cool thanks - I was just concerned if we'd accidentally regressed something as I hadn't seen this panic happening |
3 participants
This PR replaces a panic in produce_set.go with a proper error return when message set encoding fails during produce request building