[Worker] Check if worker allocator is terminated in static allocation mode #3105
Conversation
| @@ -360,6 +359,7 @@ func (k *kafka) drainOnRebalance(session sarama.ConsumerGroupSession, | |||
|
|
|||
| wg.Wait() | |||
| readyForRebalanceChan <- true | |||
| close(readyForRebalanceChan) | |||
There was a problem hiding this comment.
In that case if we are out of waiting time (in select operator), we will never close the chan because writing to the chan is blocking operation. So, I would leave it as is.
There was a problem hiding this comment.
The goroutine will still run even when the timeout has passed and the select doesn't wait for this channel.
The issue is that if timeout has passed, the function exists and closes the readyForRebalanceChan. The goroutine continues to run, and tries to write to the closed channel.
My thought was to not close the channel before the goroutine is done.
There was a problem hiding this comment.
@TomerShor Yes, but if the goroutine is still running when the timeout has passed, this change will result in the channel never being closed. We won't read from readyForRebalanceChan, leading to a zombie goroutine persisting indefinitely.
Currently, it's possible to attempt writing to a closed channel, causing a panic and the goroutine to exit. To address this issue properly and prevent the panic, we should notify the goroutine from the main function body that the timeout has passed, and there's no need to write anything to the channel. But as for me, panicing in this goroutine is not a big issue
There was a problem hiding this comment.
@rokatyy We already have a recover for that case in this goroutine 🤦
I will revert this change.
| @@ -116,13 +119,18 @@ func (s *singleton) SignalDraining() error { | |||
| } | |||
|
|
|||
| func (s *singleton) SignalTermination() error { | |||
| return s.worker.Drain() | |||
| s.isTerminated = true | |||
Followup to #3092 , where we are blocking worker allocation if a the workers are terminated:
drainOnRebalance- close thereadyForRebalanceChanafter sending a value on it. This will help cases where themaxWaitHandlerDuringRebalancetimes out before the drain handler finishes, and then the channel is closed before the drain invoking goroutine passes a value on it.