docstr: Retrieve the low and high watermark offsets from the Kafka consumer
  
  Returns
  -------
  Tuple with a [low, high] value
  
  Examples
  --------
  >>> from custream import kafka
  >>> kafka_configs = {
  ... "metadata.broker.list": "localhost:9092",
  ... "enable.partition.eof": "true",
  ... "group.id": "groupid",
  ... "auto.offset.reset": "earliest",
  ... "enable.auto.commit": "false"
  ... }
  >>> consumer = kafka.KafkaHandle(kafka_configs,
  ... topics=["kafka-topic"], partitions=[0]))
  >>> low, high = consumer.get_watermark_offsets("kafka-topic", 0)
docstr_tok: ["Retrieve","the","low","and","high","watermark","offsets","from","the","Kafka","consumer","Returns","--","--","--","-","Tuple","with","a","[","low",",","high","]","value","Examples","--","--","--","--",">",">",">","from","custream","import","kafka",">",">",">","kafka_configs","=","{","...","``","metadata.broker.list","''",":","``","localhost:9092","''",",","...","``","enable.partition.eof","''",":","``","true","''",",","...","``","group.id","''",":","``","groupid","''",",","...","``","auto.offset.reset","''",":","``","earliest","''",",","...","``","enable.auto.commit","''",":","``","false","''","...","}",">",">",">","consumer","=","kafka.KafkaHandle","(","kafka_configs",",","...","topics=","[","``","kafka-topic","''","]",",","partitions=","[","0","]",")",")",">",">",">","low",",","high","=","consumer.get_watermark_offsets","(","``","kafka-topic","''",",","0",")"]
code: def get_watermark_offsets(
      self, partition, timeout=10000, cached=False
  ):
      """
      Retrieve the low and high watermark offsets from the Kafka consumer
  
      Returns
      -------
      Tuple with a [low, high] value
  
      Examples
      --------
      >>> from custream import kafka
      >>> kafka_configs = {
      ... "metadata.broker.list": "localhost:9092",
      ... "enable.partition.eof": "true",
      ... "group.id": "groupid",
      ... "auto.offset.reset": "earliest",
      ... "enable.auto.commit": "false"
      ... }
      >>> consumer = kafka.KafkaHandle(kafka_configs,
      ... topics=["kafka-topic"], partitions=[0]))
      >>> low, high = consumer.get_watermark_offsets("kafka-topic", 0)
      """
  
      offsets = ()
  
      try:
          offsets = self.kafka_meta_client.get_watermark_offset(
              topic=partition.topic.encode(),
              partition=partition.partition,
              timeout=timeout,
              cached=cached,
          )
      except RuntimeError:
          raise RuntimeError("Unable to connect to Kafka broker")
  
      if len(offsets) != 2:
          raise RuntimeError(
              f"Multiple watermark offsets encountered. "
              f"Only 2 were expected and {len(offsets)} encountered"
          )
  
      if offsets[b"low"] < 0:
          offsets[b"low"] = 0
  
      if offsets[b"high"] < 0:
          offsets[b"high"] = 0
  
      return offsets[b"low"], offsets[b"high"]
  
file_path: cudf/python/custreamz/custreamz/kafka.py
imports: import confluent_kafka
  from cudf_kafka._lib.kafka import KafkaDatasource
  import cudf
label_desc: Use image node_id 4 for calling the Consumer obj's underlying member method code with example usage: obj.get_watermark_offsets(partition, timeout, cached) and returns: offsets,  offsets