code: def exec(self, **kwargs) -> Iterator[Batch]:
      assert (
          "input_queue" in kwargs
      ), "Invalid ray execution. No input_queue found"
      input_queue = kwargs["input_queue"]
  
      while True:
          next_item = input_queue.get(block=True)
          if next_item is StageCompleteSignal:
              # Stop signal is put back to input queue again
              # to ensure it is propagated to all ray parallel
              # actors.
              input_queue.put(StageCompleteSignal)
              break
          elif isinstance(next_item, ExecutorError):
              input_queue.put(next_item)
              raise next_item
          else:
              yield next_item
  
file_path: evadb/evadb/executor/exchange_executor.py
imports: from typing import Iterator
  from evadb.database import EvaDBDatabase
  from evadb.executor.abstract_executor import AbstractExecutor
  from evadb.executor.executor_utils import ExecutorError
  from evadb.executor.ray_utils import StageCompleteSignal, ray_parallel, ray_pull, ray_wait_and_alert
  from evadb.models.storage.batch import Batch
  from evadb.plan_nodes.exchange_plan import ExchangePlan
label_desc: Use image node_id 2 for calling the QueueReaderExecutor obj's underlying member method code with example usage: obj.exec() without return types