Fix deadlock condition

There was a corner case where a fetch thread will wait to add messages
to the queue, but the consumer has stopped consuming. It won't exit until it's
able to add all messages to the queue. In the case of a slow or stopped
consumer this will delay rebalancing because of the need to join
that thread.

The fix for now is to add a _stopped property to the partition so
the thread will know to exit. Then we empty the queue to force it to
exit instead of waiting for the next message to be consumed.

This is fairly hacky and should be replaced with a long-lived thread that
can be interrupted. However, this needs to get working now. The better
solution will be revisited later.