code: def test_dynamic_generator_distributed(
      ray_start_cluster, num_returns_type
  ):
      cluster = ray_start_cluster
      # Head node with no resources.
      cluster.add_node(num_cpus=0)
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=1)
      cluster.wait_for_nodes()
  
      @ray.remote(num_returns=num_returns_type)
      def dynamic_generator(num_returns):
          for i in range(num_returns):
              yield np.ones(1_000_000, dtype=np.int8) * i
              time.sleep(0.1)
  
      gen = ray.get(dynamic_generator.remote(3))
      for i, ref in enumerate(gen):
          # Check that we can fetch the values from a different node.
          assert ray.get(ref)[0] == i
  
imports: import pytest
  import numpy
  import sys
  import time
  import gc
  from unittest.mock import Mock
  import ray
  from ray.util.client.ray_client_helpers import ray_start_client_server_for_address
  from ray._private.client_mode_hook import enable_client_mode
  from ray.tests.conftest import call_ray_start_context
label_desc: Use image node_id 7 for calling a global function with example usage: test_dynamic_generator_distributed(ray_start_cluster, num_returns_type) without return types