code: def test_backpressure_from_output(
      ray_start_10_cpus_shared, restore_data_context
  ):
      # Here we set the memory limit low enough so the output getting blocked will
      # actually stall execution.
  
      @ray.remote
      class Counter:
          def __init__(self):
              self.i = 0
  
          def inc(self):
              self.i += 1
  
          def get(self):
              return self.i
  
      counter = Counter.remote()
  
      def func(x):
          ray.get(counter.inc.remote())
          return x
  
      ctx = DataContext.get_current()
      ctx.use_streaming_executor = True
      ctx.execution_options.resource_limits.object_store_memory = 10000
  
      # Only take the first item from the iterator.
      ds = ray.data.range(100000, parallelism=100).map_batches(
          func, batch_size=None
      )
      it = iter(ds.iter_batches(batch_size=None))
      next(it)
      time.sleep(
          3
      )  # Pause a little so anything that would be executed runs.
      num_finished = ray.get(counter.get.remote())
      assert num_finished < 20, num_finished
      # Check intermediate stats reporting.
      stats = ds.stats()
      assert "100 tasks executed" not in stats, stats
  
      # Check we can get the rest.
      for rest in it:
          pass
      assert ray.get(counter.get.remote()) == 100
      # Check final stats reporting.
      stats = ds.stats()
      assert "100 tasks executed" in stats, stats
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 12 for calling a global function with example usage: test_backpressure_from_output(ray_start_10_cpus_shared, restore_data_context) and returns: x, self