code: def test_concurrent_callable_classes(shutdown_only):
      """Test that concurrenct actor pool runs user UDF in a separate thread."""
      ray.init(num_cpus=2)
      ds = ray.data.range(10, parallelism=10)
  
      class StatefulFn:
          def __call__(self, x):
              thread_id = threading.get_ident()
              assert (
                  threading.current_thread()
                  is not threading.main_thread()
              )
              return {"tid": np.array([thread_id])}
  
      thread_ids = extract_values(
          "tid",
          ds.map_batches(
              StatefulFn, concurrency=1, max_concurrency=2
          ).take_all(),
      )
      # Make sure user's UDF is not running concurrently.
      assert len(set(thread_ids)) == 1
  
      class ErrorFn:
          def __call__(self, x):
              raise ValueError
  
      with pytest.raises(ValueError):
          ds.map_batches(
              ErrorFn, concurrency=1, max_concurrency=2
          ).take_all()
  
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 3 for calling a global function with example usage: test_concurrent_callable_classes(shutdown_only) and returns: dict