code: def get_watermark_for_topic(
      kafka_client,
      topic,
  ):
      
      # Refresh client metadata. We do not use the topic list, because we
      # don't want to accidentally create the topic if it does not exist.
      # If Kafka is unavailable, let's retry loading client metadata
      try:
          kafka_client.load_metadata_for_topics()
      except KafkaUnavailableError:
          kafka_client.load_metadata_for_topics()
  
      watermarks = get_topics_watermarks(
          kafka_client, [topic]
      )
      return watermarks
docs: This method:
          * refreshes metadata for the kafka client
          * fetches watermarks
  
      :param kafka_client: KafkaToolClient instance
      :param topic: the topic
      :returns: dict <topic>: [ConsumerPartitionOffsets]