code: def test_pulsar_write_keyed_messages():
      """Test writing keyed messages to a Pulsar topic with PulsarWriter"""
  
      topic = "test-write-keyed-messages"
      writer = tfio.experimental.streaming.PulsarWriter(
          service_url="pulsar://localhost:6650", topic=topic
      )
      # 1. Write 10 keyed messages, the key set is 0,1,2,0,1,2,...
      for i in range(10):
          value = "msg-" + str(i)
          key = str(i % 3)
          writer.write(value=value, key=key)
      writer.flush()
  
      # 2. Consume messages and verify
      dataset = tfio.experimental.streaming.PulsarIODataset(
          service_url="pulsar://localhost:6650",
          topic=topic,
          subscription="subscription-0",
          timeout=default_pulsar_timeout,
      )
      kv = dict()
      for msg, key in dataset:
          kv.setdefault(key.numpy().decode(), []).append(msg.numpy())
      assert kv["0"] == [
          ("msg-" + str(i)).encode() for i in range(0, 10, 3)
      ]
      assert kv["1"] == [
          ("msg-" + str(i)).encode() for i in range(1, 10, 3)
      ]
      assert kv["2"] == [
          ("msg-" + str(i)).encode() for i in range(2, 10, 3)
      ]
  
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 6 for calling a global function with example usage: test_pulsar_write_keyed_messages() without return types