code: def test_dynamic_empty_generator_reconstruction_nondeterministic(
      ray_start_cluster, num_returns_type
  ):
      config = {
          "health_check_failure_threshold": 10,
          "health_check_period_ms": 100,
          "health_check_initial_delay_ms": 0,
          "max_direct_call_object_size": 100,
          "task_retry_delay_ms": 100,
          "object_timeout_milliseconds": 200,
          "fetch_warn_timeout_milliseconds": 1000,
      }
      cluster = ray_start_cluster
      # Head node with no resources.
      cluster.add_node(
          num_cpus=0,
          _system_config=config,
          enable_object_reconstruction=True,
          resources={"head": 1},
      )
      ray.init(address=cluster.address)
      # Node to place the initial object.
      node_to_kill = cluster.add_node(
          num_cpus=1, object_store_memory=10**8
      )
      cluster.wait_for_nodes()
  
      @ray.remote(num_cpus=0, resources={"head": 1})
      class ExecutionCounter:
          def __init__(self):
              self.count = 0
  
          def inc(self):
              self.count += 1
              return self.count
  
          def get_count(self):
              return self.count
  
      @ray.remote(num_returns=num_returns_type)
      def maybe_empty_generator(exec_counter):
          if ray.get(exec_counter.inc.remote()) > 1:
              for i in range(3):
                  yield np.ones(1_000_000, dtype=np.int8) * i
  
      @ray.remote
      def check(empty_generator):
          return len(list(empty_generator)) == 0
  
      exec_counter = ExecutionCounter.remote()
      gen = maybe_empty_generator.remote(exec_counter)
      gen = ray.get(gen)
      refs = list(gen)
      assert ray.get(check.remote(refs))
      cluster.remove_node(node_to_kill, allow_graceful=False)
      node_to_kill = cluster.add_node(
          num_cpus=1, object_store_memory=10**8
      )
      assert ray.get(check.remote(refs))
  
      # We should never reconstruct an empty generator.
      assert ray.get(exec_counter.get_count.remote()) == 1
  
      del gen, refs, exec_counter
      assert_no_leak()
  
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 11 for calling a global function with example usage: test_dynamic_empty_generator_reconstruction_nondeterministic(ray_start_cluster, num_returns_type) and returns: unknown, self, self