If the consumer has stopped running, break out of blocking consume().
If the consumer has stopped running, break out of blocking consume().

When using a BalancedConsumer with consumer_timeout_ms=-1 and block=True (that is, full blocking consume() calls enabled), a rebalance (and thus a reallocation of the internal SimpleConsumer) caused the timeout loop in SimpleConsumer.consume() to spin infinitely. This caused execution in the caller of consume() to halt, resulting in a 'frozen' consumer at the application level.
This commit addresses this issue by ensuring that the loop is broken if the consumer has stopped running (which happens during a rebalance).

One side effect of this commit is that consumer applications that call BalancedConsumer.consume() with block=True and consumer_timeout_ms=-1 will now have to handle that function returning None messages. Previously, it didn't need to, but this commit means that even the fully-blocking mode will occasionally return None during a rebalance.

This commit addresses ticket #164

sammerry/pykafka,wikimedia/operations-debs-python-pykafka,thedrow/samsa,benauthor/pykafka,sammerry/pykafka,wikimedia/operations-debs-python-pykafka,benauthor/pykafka,sontek/pykafka,fortime/pykafka,aeroevan/pykafka,thedrow/samsa,jofusa/pykafka,fortime/pykafka,yungchin/pykafka,appsoma/pykafka,tempbottle/pykafka,vortec/pykafka,wikimedia/operations-debs-python-pykafka,sontek/pykafka,tempbottle/pykafka,appsoma/pykafka,benauthor/pykafka,jofusa/pykafka,thedrow/samsa,vortec/pykafka,aeroevan/pykafka,yungchin/pykafka