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

Author: dpkp

kafka/consumer/fetcher.py

@@ -626,9 +626,12 @@ def _handle_offset_response(self, future, response):
 
     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