openstack/deb-python-kafka
Code Issues Proposed changes

Pass consumer timeout to client.poll() in iterator; check timeout before iterating fetcher

Browse Source
This commit is contained in:
Dana Powers
2016-01-10 19:56:00 -08:00
parent b86daafeaf
commit ecb4d49c06
1 changed files with 4 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
kafka/consumer/group.py
View File

@@ -628,11 +628,14 @@ class KafkaConsumer(six.Iterator):
# init any new fetches (won't resend pending fetches) # init any new fetches (won't resend pending fetches)
self._fetcher.init_fetches() self._fetcher.init_fetches()
self._client.poll() self._client.poll(
max(0, self._consumer_timeout - time.time()) * 1000)
timeout_at = min(self._consumer_timeout, timeout_at = min(self._consumer_timeout,
self._client._delayed_tasks.next_at() + time.time(), self._client._delayed_tasks.next_at() + time.time(),
self._client.cluster.ttl() / 1000.0 + time.time()) self._client.cluster.ttl() / 1000.0 + time.time())
if time.time() > timeout_at:
continue
for msg in self._fetcher: for msg in self._fetcher:
yield msg yield msg
if time.time() > timeout_at: if time.time() > timeout_at: