Reduce internal client poll timeout for consumer interator interface

Author: dpkp

kafka/consumer/group.py

@@ -1070,8 +1070,7 @@ def _message_generator(self):
                 self._update_fetch_positions(partitions)
 
             poll_ms = 1000 * (self._consumer_timeout - time.time())
-            if not self._fetcher.in_flight_fetches():
-                poll_ms = min(poll_ms, self.config['reconnect_backoff_ms'])
+            poll_ms = min(poll_ms, self.config['retry_backoff_ms'])
             self._client.poll(timeout_ms=poll_ms)
 
             # after the long poll, we should check whether the group needs to rebalance