class_docstr: Keep the ownership and manage the workflow output.
code: def validate_init_options(
      self,
      max_running_workflows: Optional[int],
      max_pending_workflows: Optional[int],
  ):
      if (
          max_running_workflows is not None
          and max_running_workflows != self._max_running_workflows
      ) or (
          max_pending_workflows is not None
          and max_pending_workflows != self._max_pending_workflows
      ):
          raise ValueError(
              "The workflow init is called again but the init options"
              "does not match the original ones. Original options: "
              f"max_running_workflows={self._max_running_workflows} "
              f"max_pending_workflows={self._max_pending_workflows}; "
              f"New options: max_running_workflows={max_running_workflows} "
              f"max_pending_workflows={max_pending_workflows}."
          )
  
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 2 for calling the WorkflowManagementActor obj's underlying member method code with example usage: obj.validate_init_options(max_running_workflows, max_pending_workflows) without return types