code: def _retry_failed_task(
      self, workflow_id: str, failed_task_id: TaskID, exc: Exception
  ) -> bool:
      state = self._state
      is_application_error = isinstance(exc, RayTaskError)
      options = state.tasks[failed_task_id].options
      if not is_application_error or options.retry_exceptions:
          if state.task_retries[failed_task_id] < options.max_retries:
              state.task_retries[failed_task_id] += 1
              logger.info(
                  f"Retry [{workflow_id}@{failed_task_id}] "
                  f"({state.task_retries[failed_task_id]}/{options.max_retries})"
              )
              state.construct_scheduling_plan(failed_task_id)
              return True
      return False
  
imports: from typing import Dict, List, Iterator, Optional, Tuple, TYPE_CHECKING
  import asyncio
  import logging
  import time
  from collections import defaultdict
  import ray
  from ray.exceptions import RayTaskError, RayError
  from ray.workflow.common import WorkflowRef, WorkflowExecutionMetadata, WorkflowStatus, TaskID
  from ray.workflow.exceptions import WorkflowCancellationError, WorkflowExecutionError
  from ray.workflow.task_executor import get_task_executor, _BakedWorkflowInputs
  from ray.workflow.workflow_state import WorkflowExecutionState, TaskExecutionMetadata, Task
label_desc: Use image node_id 11 for calling the WorkflowExecutor obj's underlying member method code with example usage: obj._retry_failed_task(workflow_id, failed_task_id, exc) and returns: False, True