code: def test_pulsar_resubscribe():
      """Test resubscribing the same topic.
  
      If a topic is resubscribed with an existed subscription, the consumer will continue
      consuming from the last position.
  
      NOTE: This test must be run after `test_pulsar_simple_messages`.
      """
  
      topic = "test"
      writer = tfio.experimental.streaming.PulsarWriter(
          service_url="pulsar://localhost:6650", topic=topic
      )
      # 1. Append new messages to topic
      for i in range(6, 10):
          writer.write("D" + str(i))
      writer.flush()
  
      # 2. Use the same subscription with `test_pulsar_simple_messages` to continue consuming
      dataset = tfio.experimental.streaming.PulsarIODataset(
          service_url="pulsar://localhost:6650",
          topic=topic,
          subscription="subscription-0",
          timeout=default_pulsar_timeout,
      )
      assert np.all(
          [k.numpy() for (k, _) in dataset]
          == [("D" + str(i)).encode() for i in range(6, 10)]
      )
  
      # 3. Use another subscription to consume messages from beginning
      dataset = tfio.experimental.streaming.PulsarIODataset(
          service_url="pulsar://localhost:6650",
          topic=topic,
          subscription="subscription-1",
          timeout=default_pulsar_timeout,
      )
      assert np.all(
          [k.numpy() for (k, _) in dataset]
          == [("D" + str(i)).encode() for i in range(10)]
      )
  
file_path: io/tests/test_pulsar.py
imports: import sys
  import numpy
  import tensorflow
  import tensorflow_io
  import pytest
label_desc: Use image node_id 3 for calling a global function with example usage: test_pulsar_resubscribe() without return types