From b5ece5270174e7d767cb79d897c267c1fe2e47f5 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Mon, 30 Sep 2019 10:21:31 -0700 Subject: Follow up to PR 1782 -- fix tests --- kafka/consumer/fetcher.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'kafka') 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: -- cgit v1.2.1