class_docstr: Keep the ownership and manage the workflow output.
code: def __init__(
      self, max_running_workflows: int, max_pending_workflows: int
  ):
      self._workflow_executors: Dict[str, WorkflowExecutor] = {}
  
      self._max_running_workflows: int = max_running_workflows
      self._max_pending_workflows: int = max_pending_workflows
  
      # 0 means infinite for queue
      self._workflow_queue = queue.Queue(
          max_pending_workflows if max_pending_workflows != -1 else 0
      )
  
      self._running_workflows: Set[str] = set()
      self._queued_workflows: Dict[str, asyncio.Future] = {}
      # TODO(suquark): We do not cleanup "_executed_workflows" because we need to
      #  know if users are running the same workflow again long after a workflow
      #  completes. One possible alternative solution is to check the workflow
      #  status in the storage.
      self._executed_workflows: Set[str] = set()
  
imports: import asyncio
  import logging
  import queue
  from typing import Dict, List, Set, Optional, TYPE_CHECKING
  import ray
  from ray.workflow import common
  from ray.workflow.common import WorkflowStatus, TaskID
  from ray.workflow import workflow_state_from_storage
  from ray.workflow import workflow_context
  from ray.workflow import workflow_storage
  from ray.workflow.exceptions import WorkflowCancellationError, WorkflowNotFoundError, WorkflowNotResumableError, WorkflowStillActiveError
  from ray.workflow.workflow_executor import WorkflowExecutor
  from ray.workflow.workflow_state import WorkflowExecutionState
  from ray.workflow.workflow_context import WorkflowTaskContext
label_desc: Use image node_id 1 to create a new WorkflowManagementActor object  with example: obj = WorkflowManagementActor(max_running_workflows, max_pending_workflows)