code: def test_actor_pool_strategy_apply_interrupt(shutdown_only):
      """Test that _apply kills the actor pool if an interrupt is raised."""
      ray.shutdown()
  
      ray.init(include_dashboard=False, num_cpus=1)
  
      cpus = ray.available_resources()["CPU"]
      ds = ray.data.range(5, parallelism=5)
      aps = ray.data.ActorPoolStrategy(max_size=5)
      blocks = ds._plan.execute()
  
      # Start some actors, the first one sends a SIGINT, emulating a KeyboardInterrupt
      def test_func(block):
          for i, _ in enumerate(
              BlockAccessor.for_block(block).iter_rows()
          ):
              if i == 0:
                  os.kill(os.getpid(), signal.SIGINT)
              else:
                  time.sleep(1000)
                  return block
  
      # No need to test ActorPoolStrategy in new execution backend.
      if not DataContext.get_current().new_execution_backend:
          with pytest.raises(ray.exceptions.RayTaskError):
              aps._apply(test_func, {}, blocks, False)
  
      # Check that all actors have been killed by counting the available CPUs
      wait_for_condition(
          lambda: (ray.available_resources().get("CPU", 0) == cpus)
      )
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 25 for calling a global function with example usage: test_actor_pool_strategy_apply_interrupt(shutdown_only) and returns: block