Use local copies in Fetcher._fetchable_partitions to avoid mutation errors (#1400)

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