Empty buffer after preparing _flush_endpoint in BufferedConsumer

[kevinyu]

An asynchronous buffered consumer will duplicate messages if it is flushed while an earlier call to _flush_endpoint is being processed asynchronously. Discovered this while using mixpanel-python-async.

By emptying the buffer before iterating over the messages, subsequent calls to _flush_endpoint while async call(s) are already in progress will not duplicate messages.

[zakj]

Thanks for this, @kevinyu! Could you help me understand how you're triggering this problem? We don't maintain mixpanel-python-async, but as I understand their AsyncBufferedConsumer it uses a lock to ensure only one thread can be flushing at once.

[kevinyu]

Hi @zakj. I was a bit mistaken in my description of the source of the duplication. You are totally right about the lock preventing two async flushes at the same time.

I actually encountered this this by making a synchronous flush (i.e. consumer.flush(async=False) using AsyncBufferedConsumer) while an asynchronous flushing thread was already in progress. Perhaps this is an issue more suited for the mixpanel-python-async project then.

Also as a side note, I just realized that my change would alter the behavior of the consumer if a MixpanelException is raised, by not preserving the buffer's contents in the event of the exception.

[zakj]

Ahh, that makes total sense. While I hate to punt you away to another repo, I think in this case you're right that the fix belongs in AsyncBufferedConsumer. I'm going to close this PR for now, but do let me know if you have trouble getting your problem solved over there. Thanks again!