diff options
author | Dana Powers <dana.powers@rd.io> | 2018-02-28 18:12:23 -0800 |
---|---|---|
committer | Dana Powers <dana.powers@rd.io> | 2018-02-28 18:12:23 -0800 |
commit | 091f4c5dd161d51678401600766dcaeac2ad194d (patch) | |
tree | a43db8f4f73fcf0c77c5ae875aa4997217bb6e60 | |
parent | 4cadaafb24c2bdad475a68e3df5a4e19ce043ce7 (diff) | |
download | kafka-python-issue_1399.tar.gz |
Use local copies in Fetcher._fetchable_partitions to avoid mutation errorsissue_1399
-rw-r--r-- | kafka/consumer/fetcher.py | 9 |
1 files changed, 6 insertions, 3 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 4f2a543..ea7d5d8 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -626,9 +626,12 @@ class Fetcher(six.Iterator): def _fetchable_partitions(self): fetchable = self._subscriptions.fetchable_partitions() - if self._next_partition_records: - fetchable.discard(self._next_partition_records.topic_partition) - for fetch in self._completed_fetches: + # do not fetch a partition if we have a pending fetch response to process + current = self._next_partition_records + pending = copy.copy(self._completed_fetches) + if current: + fetchable.discard(current.topic_partition) + for fetch in pending: fetchable.discard(fetch.topic_partition) return fetchable |