class_docstr: PulsarIODataset
docstr: Creates a `PulsarIODataset` from pulsar server with a subscription
  
  Args:
    service_url: A `tf.string` tensor containing the service url of pulsar broker.
      For example: "pulsar://localhost:6650".
    topic: A `tf.string` tensor containing the topic name.
    subscription: A `tf.string` tensor containing the subscription name.
    ack_grouping_time: A `tf.int64` tensor containing the ack grouping time.
      If it's non-negative, each time a message was received, the consumer would add it to
      a batch and acknowledge all pending messages later. `ack_grouping_time` is the interval
      between two acknowledgements in milliseconds.
      If it's negative, the message would be acknowledged immediately.
      Default: -1
    timeout: A `tf.int64` tensor containing the timeout in milliseconds. If no messages
      were received after `timeout` milliseconds, we can treat it as no more messages.
      `timeout` must be positive.
    poll_timeout: A `tf.int64` tensor containing the poll timeout in milliseconds. The
      pulsar consumer would try to receive a single message with the poll timeout, if no
      message was received, it would try again until `timeout` exceeds.
      `poll_timeout` must be positive and not larger than `timeout`.
      Default: 100
docstr_tok: ["Creates","a","`","PulsarIODataset","`","from","pulsar","server","with","a","subscription","Args",":","service_url",":","A","`","tf.string","`","tensor","containing","the","service","url","of","pulsar","broker",".","For","example",":","``","pulsar",":","\/\/localhost:6650","''",".","topic",":","A","`","tf.string","`","tensor","containing","the","topic","name",".","subscription",":","A","`","tf.string","`","tensor","containing","the","subscription","name",".","ack_grouping_time",":","A","`","tf.int64","`","tensor","containing","the","ack","grouping","time",".","If","it","'s","non-negative",",","each","time","a","message","was","received",",","the","consumer","would","add","it","to","a","batch","and","acknowledge","all","pending","messages","later",".","`","ack_grouping_time","`","is","the","interval","between","two","acknowledgements","in","milliseconds",".","If","it","'s","negative",",","the","message","would","be","acknowledged","immediately",".","Default",":","-1","timeout",":","A","`","tf.int64","`","tensor","containing","the","timeout","in","milliseconds",".","If","no","messages","were","received","after","`","timeout","`","milliseconds",",","we","can","treat","it","as","no","more","messages",".","`","timeout","`","must","be","positive",".","poll_timeout",":","A","`","tf.int64","`","tensor","containing","the","poll","timeout","in","milliseconds",".","The","pulsar","consumer","would","try","to","receive","a","single","message","with","the","poll","timeout",",","if","no","message","was","received",",","it","would","try","again","until","`","timeout","`","exceeds",".","`","poll_timeout","`","must","be","positive","and","not","larger","than","`","timeout","`",".","Default",":","100"]
code: def __init__(
      self,
      service_url,
      topic,
      subscription,
      timeout,
      ack_grouping_time=-1,
      poll_timeout=100,
  ):
      """Creates a `PulsarIODataset` from pulsar server with a subscription
  
      Args:
        service_url: A `tf.string` tensor containing the service url of pulsar broker.
          For example: "pulsar://localhost:6650".
        topic: A `tf.string` tensor containing the topic name.
        subscription: A `tf.string` tensor containing the subscription name.
        ack_grouping_time: A `tf.int64` tensor containing the ack grouping time.
          If it's non-negative, each time a message was received, the consumer would add it to
          a batch and acknowledge all pending messages later. `ack_grouping_time` is the interval
          between two acknowledgements in milliseconds.
          If it's negative, the message would be acknowledged immediately.
          Default: -1
        timeout: A `tf.int64` tensor containing the timeout in milliseconds. If no messages
          were received after `timeout` milliseconds, we can treat it as no more messages.
          `timeout` must be positive.
        poll_timeout: A `tf.int64` tensor containing the poll timeout in milliseconds. The
          pulsar consumer would try to receive a single message with the poll timeout, if no
          message was received, it would try again until `timeout` exceeds.
          `poll_timeout` must be positive and not larger than `timeout`.
          Default: 100
      """
      with tf.name_scope("PulsarIODataset"):
          if timeout <= 0:
              raise ValueError(
                  f"Invalid timeout value: {timeout}, must be > 0"
              )
  
          if poll_timeout <= 0:
              raise ValueError(
                  f"Invalid poll_timeout value: {poll_timeout}, must be > 0"
              )
  
          if poll_timeout > timeout:
              raise ValueError(
                  "Invalid poll_timeout value: {}, must be <= timeout({})".format(
                      poll_timeout, timeout
                  )
              )
  
          resource = core_ops.io_pulsar_readable_init(
              service_url, topic, subscription, ack_grouping_time
          )
          self._resource = resource
          dataset = tf.data.experimental.Counter()
          dataset = dataset.map(
              lambda i: core_ops.io_pulsar_readable_next(
                  input=self._resource,
                  timeout=timeout,
                  poll_timeout=poll_timeout,
              )
          )
          dataset = dataset.apply(
              tf.data.experimental.take_while(
                  lambda v: tf.greater(v.continue_fetch, 0)
              )
          )
          dataset = dataset.map(lambda v: (v.message, v.key))
          dataset = dataset.unbatch()
  
          self._dataset = dataset
          super().__init__(self._dataset._variant_tensor)
  
imports: import tensorflow
  from tensorflow_io.python.ops import core_ops
label_desc: Use image node_id 1 to create a new PulsarIODataset object from inherited base classes: tf.data with example: obj = PulsarIODataset(service_url, topic, subscription, timeout, ack_grouping_time, poll_timeout)