code: def test_pulsar_keyed_messages():
      """Test consuming keyed messages from a Pulsar topic with PulsarIODataset
  
      NOTE: After the pulsar standalone is setup during the testing phase, 6 messages
      are sent to the `test` topic:
  
      K0:D0, K1:D1, K0:D2, K1:D3, K0:D4, K1:D5.
      """
  
      dataset = tfio.experimental.streaming.PulsarIODataset(
          service_url="pulsar://localhost:6650",
          topic="key-test",
          subscription="subscription-0",
          timeout=default_pulsar_timeout,
      )
      kv = dict()
      for msg, key in dataset:
          kv.setdefault(key.numpy().decode(), []).append(msg.numpy())
      assert kv["K0"] == [
          ("D" + str(i)).encode() for i in range(0, 6, 2)
      ]
      assert kv["K1"] == [
          ("D" + str(i)).encode() for i in range(1, 6, 2)
      ]
  
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 2 for calling a global function with example usage: test_pulsar_keyed_messages() without return types