Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Do network connections and writes in KafkaClient.poll() #1729

Merged
merged 6 commits into from
Mar 8, 2019
Merged

Do network connections and writes in KafkaClient.poll() #1729

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3c87379
Add BrokerConnection.send_pending_requests to support async network s…
dpkp Mar 6, 2019
492b7d2
Send network requests during KafkaClient.poll() rather than in KafkaC…
dpkp Mar 6, 2019
2313286
Dont acquire lock during KafkaClient.send if node is connected / ready
dpkp Mar 6, 2019
957c62d
Move all network connection IO into KafkaClient.poll()
dpkp Mar 6, 2019
7ce6976
Fetcher should call client.poll() regardless of in-flight-request count
dpkp Mar 6, 2019
0a8914f
Remove sleep check when no partitions assigned -- no longer needed
dpkp Mar 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Remove sleep check when no partitions assigned -- no longer needed
  • Loading branch information
@dpkp @jeffwidman
dpkp authored and jeffwidman committed Mar 8, 2019
commit 0a8914ff0e9e5896b761127b82b5aac613d3e4c0
10 changes: 0 additions & 10 deletions kafka/consumer/group.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1070,16 +1070,6 @@ def _message_generator(self):
# like heartbeats, auto-commits, and metadata refreshes
timeout_at = self._next_timeout()

# Because the consumer client poll does not sleep unless blocking on
# network IO, we need to explicitly sleep when we know we are idle
# because we haven't been assigned any partitions to fetch / consume
if self._use_consumer_group() and not self.assignment():
sleep_time = max(timeout_at - time.time(), 0)
if sleep_time > 0 and not self._client.in_flight_request_count():
log.debug('No partitions assigned; sleeping for %s', sleep_time)
time.sleep(sleep_time)
continue

# Short-circuit the fetch iterator if we are already timed out
# to avoid any unintentional interaction with fetcher setup
if time.time() > timeout_at:
Expand Down