code: def advance_consumer_offsets(
      kafka_client,
      group,
      topics,
      raise_on_error=True,
  ):
      
      kafka_client.load_metadata_for_topics()
  
      return _commit_offsets_to_watermark(
          kafka_client, group, topics,
          HIGH_WATERMARK, raise_on_error,
      )
docs: Advance consumer offsets to the latest message in the topic
      partition (the high watermark).
  
      This method shall refresh the client metadata prior to updating
      the offsets.
  
      If any partition leader is not available, the request fails for all the
      other topics. This is the tradeoff of sending all topic requests in batch
      and save both in performance and Kafka load.
  
      :param kafka_client: a connected KafkaToolClient
      :param group: kafka group_id
      :param topics: topic list or dict {<topic>: [partitions]}
      :param raise_on_error: if False the method does not raise exceptions
        on missing topics/partitions. It may still fail on the request send.
      :returns: a list of errors for each partition offset update that failed.
      :rtype: list [OffsetCommitError]
      :raises:
        :py:class:`kafka_utils.util.error.UnknownTopic`: upon missing
        topics and raise_on_error=True
  
        :py:class:`kafka_utils.util.error.UnknownPartition`: upon missing
        partitions and raise_on_error=True
  
        FailedPayloadsError: upon send request error.