code: def test_retry_application_level_error_exception_filter(
      ray_start_regular,
  ):
      @ray.remote
      class Counter:
          def __init__(self):
              self.value = 0
  
          def increment(self):
              self.value += 1
              return self.value
  
      @ray.remote(max_retries=1, retry_exceptions=[CountError])
      def func(counter):
          if counter is None:
              raise ValueError()
          count = counter.increment.remote()
          if ray.get(count) == 1:
              raise CountError()
          else:
              return 2
  
      # Exception that doesn't satisfy the predicate should cause the task to immediately
      # fail.
      r0 = func.remote(None)
      with pytest.raises(ValueError):
          ray.get(r0)
  
      # Test against exceptions (CountError) that do satisfy the predicate.
      counter1 = Counter.remote()
      r1 = func.remote(counter1)
      assert ray.get(r1) == 2
  
      counter2 = Counter.remote()
      r2 = func.options(max_retries=0).remote(counter2)
      with pytest.raises(CountError):
          ray.get(r2)
  
      counter3 = Counter.remote()
      r3 = func.options(retry_exceptions=False).remote(counter3)
      with pytest.raises(CountError):
          ray.get(r3)
  
imports: import subprocess
  import sys
  import time
  import grpc
  import numpy
  import psutil
  import pytest
  from grpc._channel import _InactiveRpcError
  from ray.util.state import list_tasks
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  import ray._private.ray_constants
  import ray.experimental.internal_kv
  from ray import NodeID
  from ray._private.test_utils import SignalActor, get_error_message, init_error_pubsub, run_string_as_driver, wait_for_condition, kill_raylet
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.core.generated import gcs_service_pb2, gcs_service_pb2_grpc, node_manager_pb2, node_manager_pb2_grpc
  from ray.exceptions import LocalRayletDiedError
label_desc: Use image node_id 4 for calling a global function with example usage: test_retry_application_level_error_exception_filter(ray_start_regular) and returns: self, int