code: def get_consumer_offsets_metadata(
      kafka_client,
      group,
      topics,
      raise_on_error=True,
  ):
      
      # 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()
  
      group_offsets = get_current_consumer_offsets(
          kafka_client, group, topics, raise_on_error
      )
  
      watermarks = get_topics_watermarks(
          kafka_client, topics, raise_on_error
      )
  
      result = {}
      for topic, partitions in six.iteritems(group_offsets):
          result[topic] = [
              ConsumerPartitionOffsets(
                  topic=topic,
                  partition=partition,
                  current=group_offsets[topic][partition],
                  highmark=watermarks[topic][partition].highmark,
                  lowmark=watermarks[topic][partition].lowmark,
              ) for partition in partitions
          ]
      return result
docs: This method:
          * refreshes metadata for the kafka client
          * fetches group offsets
          * fetches watermarks
  
      :param kafka_client: KafkaToolClient instance
      :param group: group id
      :param topics: list of topics
      :param raise_on_error: if False the method ignores missing topics and
        missing partitions. It still may fail on the request send.
      :returns: dict <topic>: [ConsumerPartitionOffsets]