Fix consumer poll stuck error when no available partition

[ckyoog]

The consumer poll is stuck due to long timeout when there is no more partition for it.

Actually, in function commit_offsets_async(), passing 0 as timeout into client.poll() is the very action the official Java source code does.

[ckyoog]

The bug can be reproduced in this way:

1. create topic 'test' with only one partition
2. start a consumer with an explicit 'group_id' to consume topic 'test'
3. start another consumer with same configs to subscribe the same topic

The second consumer will be stuck (sleep almost 300 seconds) at the place where my change is, even the first consumer is kicked out due to expiration (exceeding the max_poll_interval_ms), or even the first consumer exits (close).

The matching place of Java code is
in function commitOffsetsAsync()
in class ConsumerCoordinator
in file kafka-1.0.0-src/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

In Java code, commitOffsetsAsync() calls pollNoWakeup(), which passes 0 as timeout to the subsequent calls, up to client.poll(), in which the client is class NetworkClient.

So in this context, the call to self._client.poll() in python code is just equivalent to the call to client.poll() in Java code, but the java one gets 0 as timeout whereas the python one doesn't.

[dpkp]

Thanks!