From a6130d288f84af7ffe054bdf301bc40febe07719 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 7 Mar 2018 17:07:09 -0800 Subject: Use local copies in Fetcher._fetchable_partitions to avoid mutation errors (#1400) --- kafka/consumer/fetcher.py | 9 ++++++--- 1 file 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 -- cgit v1.2.1