code: def test_dynamic_generator_retry_exception(
      ray_start_regular, store_in_plasma, num_returns_type
  ):
      class CustomException(Exception):
          pass
  
      @ray.remote(num_cpus=0)
      class ExecutionCounter:
          def __init__(self):
              self.count = 0
  
          def inc(self):
              self.count += 1
              return self.count
  
          def get_count(self):
              return self.count
  
          def reset(self):
              self.count = 0
  
      @ray.remote(max_retries=1)
      def generator(num_returns, store_in_plasma, counter):
          for i in range(num_returns):
              if store_in_plasma:
                  yield np.ones(1_000_000, dtype=np.int8) * i
              else:
                  yield [i]
  
              # Fail on first execution, succeed on next.
              if ray.get(counter.inc.remote()) == 1:
                  raise CustomException("error")
  
      counter = ExecutionCounter.remote()
      dynamic_ref = generator.options(
          num_returns=num_returns_type
      ).remote(3, store_in_plasma, counter)
      ref1, ref2 = ray.get(dynamic_ref)
      ray.get(ref1)
      with pytest.raises(ray.exceptions.RayTaskError):
          ray.get(ref2)
  
      ray.get(counter.reset.remote())
      dynamic_ref = generator.options(
          num_returns=num_returns_type,
          retry_exceptions=[CustomException],
      ).remote(3, store_in_plasma, counter)
      for i, ref in enumerate(ray.get(dynamic_ref)):
          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 5 for calling a global function with example usage: test_dynamic_generator_retry_exception(ray_start_regular, store_in_plasma, num_returns_type) and returns: self, self