Avoid KeyError when filtering fetchable partitions

1 files changed, 4 insertions, 2 deletions

diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index debe86b..0c3b84c 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -638,9 +638,11 @@ class Fetcher(six.Iterator):
     def _fetchable_partitions(self):
         fetchable = self._subscriptions.fetchable_partitions()
         if self._next_partition_records:
-            fetchable.remove(self._next_partition_records.topic_partition)
+            if self._next_partition_records.topic_partition in fetchable:
+                fetchable.remove(self._next_partition_records.topic_partition)
         for fetch in self._completed_fetches:
-            fetchable.remove(fetch.topic_partition)
+            if fetch.topic_partition in fetchable:
+                fetchable.remove(fetch.topic_partition)
         return fetchable
 
     def _create_fetch_requests(self):