From 091f4c5dd161d51678401600766dcaeac2ad194d Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 28 Feb 2018 18:12:23 -0800 Subject: Use local copies in Fetcher._fetchable_partitions to avoid mutation errors --- 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