From 229af251b97327ced11b8c3bdc687a3d383ac816 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 14 Mar 2025 16:15:51 -0700 Subject: [PATCH] Log warning when attempting to list offsets for unknown topic/partition --- kafka/consumer/fetcher.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 2179e19fc..e5ae64c91 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -538,9 +538,13 @@ def _send_list_offsets_requests(self, timestamps): for partition, timestamp in six.iteritems(timestamps): node_id = self._client.cluster.leader_for_partition(partition) if node_id is None: + if partition.topic not in self._client.cluster.topics(): + log.warning("Could not lookup offsets for partition %s since no metadata is available for topic. " + "Wait for metadata refresh and try again", partition) + else: + log.warning("Could not lookup offsets for partition %s since no metadata is available for it. " + "Wait for metadata refresh and try again", partition) self._client.add_topic(partition.topic) - log.debug("Partition %s is unknown for fetching offset," - " wait for metadata refresh", partition) return Future().failure(Errors.StaleMetadata(partition)) elif node_id == -1: log.debug("Leader for partition %s unavailable for fetching "