code: def test_finalize_fn_uses_single_thread(ray_start_regular_shared):
      """Tests that finalize_fn is not run with multiple threads."""
      block_refs_iter = itertools.starmap(
          lambda block, metadata: (ray.put(block), metadata),
          block_generator(num_blocks=20, num_rows=2),
      )
  
      q = queue.Queue()
      semaphore = threading.Semaphore(value=1)
  
      def finalize_enforce_single_thread(batch):
          already_acquired = not semaphore.acquire(blocking=False)
          if already_acquired:
              e = AssertionError(
                  "finalize_fn is being run concurrently."
              )
              q.put(e, block=True)
          semaphore.release()
          return batch
  
      # Test that finalize_fn is called in a single thread,
      # even if prefetch_batches is set.
      output_batches = iter_batches(
          block_refs_iter,
          dataset_tag="dataset",
          collate_fn=lambda batch: batch,
          finalize_fn=finalize_enforce_single_thread,
          prefetch_batches=4,
      )
  
      # Force execution of the iterator.
      # This step should not raise an exception.
      list(output_batches)
  
      try:
          e = q.get(block=False, timeout=0.1)
          raise e
      except queue.Empty:
          pass
  
file_path: ray/python/ray/data/tests/block_batching/test_iter_batches.py
imports: import itertools
  import queue
  import threading
  import time
  from typing import Iterator, List, Tuple
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher
  from ray.data._internal.block_batching.iter_batches import iter_batches, prefetch_batches_locally, restore_original_order
  from ray.data.block import Block, BlockMetadata
label_desc: Use image node_id 4 for calling a global function with example usage: test_finalize_fn_uses_single_thread(ray_start_regular_shared) and returns: batch