diff options
| -rw-r--r-- | kafka/consumer/fetcher.py | 9 |
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 |