code: def test_publish_and_subscribe_logs(ray_start_regular):
      address_info = ray_start_regular
      gcs_server_addr = address_info["gcs_address"]
  
      subscriber = ray._raylet.GcsLogSubscriber(address=gcs_server_addr)
      subscriber.subscribe()
  
      publisher = ray._raylet.GcsPublisher(address=gcs_server_addr)
      log_batch = {
          "ip": "127.0.0.1",
          "pid": 1234,
          "job": "0001",
          "is_err": False,
          "lines": ["line 1", "line 2"],
          "actor_name": "test actor",
          "task_name": "test task",
      }
      publisher.publish_logs(log_batch)
  
      # PID is treated as string.
      log_batch["pid"] = "1234"
      assert subscriber.poll() == log_batch
  
      subscriber.close()
  
file_path: ray/python/ray/tests/test_gcs_pubsub.py
imports: import asyncio
  import sys
  import threading
  import re
  import ray
  from ray._private.gcs_pubsub import GcsAioPublisher, GcsAioErrorSubscriber, GcsAioLogSubscriber, GcsAioResourceUsageSubscriber
  from ray.core.generated.gcs_pb2 import ErrorTableData
  import pytest
label_desc: Use image node_id 2 for calling a global function with example usage: test_publish_and_subscribe_logs(ray_start_regular) without return types