code: def test_actor_pool_strategy_bundles_to_max_actors(shutdown_only):
      """Tests that blocks are bundled up to the specified max number of actors."""
  
      class UDFClass:
          def __call__(self, x):
              return x
  
      max_size = 2
      ds = (
          ray.data.range(10, parallelism=10)
          .map_batches(UDFClass, batch_size=None, concurrency=max_size)
          .materialize()
      )
  
      # TODO(https://github.com/ray-project/ray/issues/31723): implement the feature
      # of capping bundle size by actor pool size, and then re-enable this test.
      if not DataContext.get_current().new_execution_backend:
          assert f"{max_size} blocks" in ds.stats()
  
      # Check batch size is still respected.
      ds = (
          ray.data.range(10, parallelism=10)
          .map_batches(UDFClass, batch_size=10, concurrency=max_size)
          .materialize()
      )
  
      assert "1 blocks" in ds.stats()
  
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 27 for calling a global function with example usage: test_actor_pool_strategy_bundles_to_max_actors(shutdown_only) and returns: x