Fix shutdown and race-condition in consumer-group example

[skidder]

The Kafka consumer groups example has two issues:

1. A race condition causes the ready channel to be reassigned before the consumer starts, leading to the main go-routine reading from a different ready channel than the one that is closed in the consumer Setup function.
2. The consumer go-routine doesn't exit cleanly when a termination signal is sent. But the termination signal handling wasn't configured because of the aforementioned race condition, so this wouldn't have worked regardless.

This PR fixes both problems by reassigning the ready channel variable only after a rebalance, and by using a cancellable context and waitgroup to ensure that the Kafka client is closed cleanly upon receiving a termination signal.

[bai]

Many thanks!

[mabrarov]

Doesn't this reassigning can still cause race condition with line 100 if consumer goroutine is fast enough? Isn't reassigning channel variable in one goroutine while receiving from / sending into a channel accessed through the same variable in another goroutine is a race condition in general?

[puellanivis]

No, the receive on line 100 is done by retrieving the channel, and then waiting for the result. Since that channel is setup on line 75, that will not cause an issue with this assignment being made to the same location. Even if this line of code runs and replaces the value, the receive on line 100 is still going to be on whatever channel was there when line 100 started waiting.

Oddly, I believe this assignment is basically ineffective as a result? Nothing else is listening on this new channel, so… 🤷‍♀️ It’s hard to reason properly about though. I’m not sure why we wouldn’t just setup the signal handler right away, and handle the consumer ready and signal handling all together…

In fact, I’m not sure why we have to wait on the consumer setting up at all?!

[mabrarov]

Thank you for quick response.

In fact, I’m not sure why we have to wait on the consumer setting up at all?!

I agree, but the fix seems to be pretty easy - refer to main...mabrarov:sarama:fix/example_consumergroup_race_condition. Easier than getting agreement on changing the execution flow (maybe example wanted to demo something with this wait).

Even if this line of code runs and replaces the value, the receive on line 100 is still going to be on whatever channel was there when line 100 started waiting.

Isn't reading from variable changed in another goroutine without synchronization considered race-condition in general?

[mabrarov]

... following https://go.dev/ref/mem#restrictions and considering Go channel under the hood should be a pointer (i.e. variable of channel type is not larger than a machine word - refer to https://go.dev/play/p/dvM6ow3Op_S) we should be OK at line 100 (Google says for Single-Word or Sub-Word Types ... Each read must observe a value actually written to that location by a concurrent or preceding write) unless line 100 executes before consumer.ready is reassigned and consumer.ready is reassigned without closing old channel (which is potentially possible). Do I understand your point correctly?

[puellanivis]

I agree, but the fix seems to be pretty easy

This is not a fix, it is a further complication—an unnecessary one as explained below—that only further complicates an example that is overly complicated to begin with.

The real “fix” should we wish to keep the wait for consumer to be ready would be to just never replace the channel. We only need the one time trigger ever anyways. Then the consumer is single threaded, so just close it under a test that it’s not already closed. Done.

Isn't reading from variable changed in another goroutine without synchronization considered race-condition in general?

No, the read happened well before the write occurred, so there’s no write-read race. Line 100 reads the pointer, and waits on the underlying channel behind that variable, at this point that variable can be modified all you want, and it will not race with the receive. The receive is already waiting on the underlying channel, NOT on the variable holding that reference.

Once the recv blocks there is no longer any possibility for a race condition, because the underlying channel has already been resolved. There’s no ongoing continuous read from that variable to maintain the wait on the receive.

[mabrarov]

No, the read happened well before the write occurred

Sorry, but I don't get why we are ensured that the read in main goroutine happens well before the write in consumer goroutine occurs? There is no happens before relationship b/w these read and write operations to guarantee the order ("potentional" word in #3674 comes from this point). Did I miss something?

[puellanivis]

So, this is subtle knowledge of how goroutines are scheduled. In that, goroutines are actually never guaranteed to run. This means that the main goroutine after queuing up the go func will immediately continue executing and enter the receive.

For there to be any sort of write-read race condition, it requires the sub-groutine to have not only started execution, but also execute its Setup, (the caller of which has a hojillion synchronizing events), a read from a closed claim.Message (which requires a call to PartitionConsumer.AsyncClose which has a synchronizing sync.Once.Do itself, but to handle while in ConsumeClaim requires a context.CancelFunc or close(channel) to have happened, both of which are synchronizing) so that ConsumeClaim returns, and a call to Cleanup whose caller has a to cancel a context, wait on a WaitGroup, then a sync.Once.Do again, and then an atomic load in the ctx.Err that returns nil. All of that would have to happen before the main goroutine hit line 100.

There is so much that the sub-goroutine has to do, with so many synchronization points where the sub-goroutine would block… meanwhile the main goroutine has nothing blocking until it has entered the receive on consumer.ready. The gap—should any technically exist—is sub-Planck-length wide. A truly pathological case of scheduling would have to happen.

P.S.: The sub-goroutine will necessarily block before the assignment and before the main goroutine enters the receive.

[mabrarov]

So, it is considered OK to rely on implementation of sarama.ConsumerGroup.Consume and respective timings (which I agree goes to sub-Planck-length wide) instead of explicit synchronization b/w main (consumer.ready read operation) and consumer (consumer.ready write operation) goroutines (which is still formally considered race-condition according to Go memory model, isn't it?). Sounds acceptable in practice, but doubtful from readibility point of view which is especially important for examples.

Got it. Thank you for confirming my suspicion.

[puellanivis]

I mean, if you want to update and simplify things which also ensures that synchronization is guaranteed, that’s fine. Like, I said, the code itself is super out-of-date by now, and could definitely be improved.

But not fixing the existing issues, and further complicating things with workarounds to close a merely theoretical race condition makes an example harder to read and understand.

Also, I wouldn’t say it’s even formally a race-condition. Sarama has tons of synchronization inside the call to Consume between calling the Setup Messages and Cleanup that ensure that this is well behaved. Relying on that internal synchronization is fine.