diff options
| author | Dana Powers <dana.powers@gmail.com> | 2019-09-30 11:18:46 -0700 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2019-09-30 11:18:46 -0700 |
| commit | 975087b4ae22ecdb63b757a5abe2e9643888b7fe (patch) | |
| tree | aead2ad0ab08b1c4c2e0a8b9804a3e634ee099e4 /kafka/consumer/fetcher.py | |
| parent | 87fb1bb48f82bcaa6c5e1a1edadab2832659801c (diff) | |
| download | kafka-python-975087b4ae22ecdb63b757a5abe2e9643888b7fe.tar.gz | |
Follow up to PR 1782 -- fix tests (#1914)
Diffstat (limited to 'kafka/consumer/fetcher.py')
| -rw-r--r-- | kafka/consumer/fetcher.py | 3 |
1 files changed, 2 insertions, 1 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index f781d4c..5434c36 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -268,6 +268,7 @@ class Fetcher(six.Iterator): start_time = time.time() remaining_ms = timeout_ms + timestamps = copy.copy(timestamps) while remaining_ms > 0: if not timestamps: return {} @@ -294,7 +295,7 @@ class Fetcher(six.Iterator): if refresh_future.succeeded() and isinstance(future.exception, Errors.StaleMetadata): log.debug("Stale metadata was raised, and we now have an updated metadata. Rechecking partition existance") unknown_partition = future.exception.args[0] # TopicPartition from StaleMetadata - if not self._client.cluster.leader_for_partition(unknown_partition): + if self._client.cluster.leader_for_partition(unknown_partition) is None: log.debug("Removed partition %s from offsets retrieval" % (unknown_partition, )) timestamps.pop(unknown_partition) else: |