class_docstr: A streaming Dataset executor.
  
  This implementation executes Dataset DAGs in a fully streamed way. It runs
  by setting up the operator topology, and then routing blocks through operators in
  a way that maximizes throughput under resource constraints.
class_docstr_tok: ["A","streaming","Dataset","executor",".","This","implementation","executes","Dataset","DAGs","in","a","fully","streamed","way",".","It","runs","by","setting","up","the","operator","topology",",","and","then","routing","blocks","through","operators","in","a","way","that","maximizes","throughput","under","resource","constraints","."]
code: def shutdown(self, execution_completed: bool = True):
      context = DataContext.get_current()
      global _num_shutdown
  
      with self._shutdown_lock:
          if not self._execution_started or self._shutdown:
              return
          logger.get_logger().debug(f"Shutting down {self}.")
          _num_shutdown += 1
          self._shutdown = True
          # Give the scheduling loop some time to finish processing.
          self.join(timeout=2.0)
          self._update_stats_metrics(
              state="FINISHED" if execution_completed else "FAILED",
              force_update=True,
          )
          # Clears metrics for this dataset so that they do
          # not persist in the grafana dashboard after execution
          StatsManager.clear_execution_metrics(
              self._dataset_tag, self._get_operator_tags()
          )
          # Freeze the stats and save it.
          self._final_stats = self._generate_stats()
          stats_summary_string = (
              self._final_stats.to_summary().to_string(
                  include_parent=False
              )
          )
          logger.get_logger(
              log_to_stdout=context.enable_auto_log_stats
          ).info(
              stats_summary_string,
          )
          # Close the progress bars from top to bottom to avoid them jumping
          # around in the console after completion.
          if self._global_info:
              self._global_info.close()
          for op, state in self._topology.items():
              op.shutdown()
              state.close_progress_bars()
          # Make request for zero resources to autoscaler for this execution.
          actor = get_or_create_autoscaling_requester_actor()
          actor.request_resources.remote({}, self._execution_id)
  
imports: import os
  import threading
  import time
  import uuid
  from typing import Dict, Iterator, List, Optional
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.autoscaling_requester import get_or_create_autoscaling_requester_actor
  from ray.data._internal.execution.backpressure_policy import BackpressurePolicy, get_backpressure_policies
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, Executor, OutputIterator, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.streaming_executor_state import DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION, AutoscalingState, OpState, Topology, TopologyResourceUsage, build_streaming_topology, process_completed_tasks, select_operator_to_run, update_operator_states
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.stats import DatasetStats, StatsManager
  from ray.data._internal.util import cluster_resources
  from ray.data.context import DataContext
label_desc: Use image node_id 4 for calling the StreamingExecutor obj's underlying member method code with example usage: obj.shutdown(execution_completed) without return types