Parsely / pykafka / 571

38%

push

travis-ci

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

2 of 2 new or added lines in 1 file covered. (100.0%)

1321 of 1843 relevant lines covered (71.68%)

0.72 hits per line