class_docstr: MetaflowTask prepares a Flow instance for execution of a single step.
class_docstr_tok: ["MetaflowTask","prepares","a","Flow","instance","for","execution","of","a","single","step","."]
code: def clone_only(
      self,
      step_name,
      run_id,
      task_id,
      clone_origin_task,
      retry_count,
      wait_only=False,
  ):
      if not clone_origin_task:
          raise MetaflowInternalError(
              "task.clone_only needs a valid clone_origin_task value."
          )
      if wait_only:
          # In this case, we are actually going to wait for the clone to be done
          # by someone else. To do this, we just get the task_datastore in "r" mode
          while True:
              try:
                  ds = self.flow_datastore.get_task_datastore(
                      run_id, step_name, task_id
                  )
                  if not ds["_task_ok"]:
                      raise MetaflowInternalError(
                          "Externally cloned task did not succeed"
                      )
                  break
              except DataException:
                  # No need to get fancy with the sleep here.
                  time.sleep(5)
          return
      # If we actually have to do the clone ourselves, proceed...
      # 1. initialize output datastore
      output = self.flow_datastore.get_task_datastore(
          run_id, step_name, task_id, attempt=0, mode="w"
      )
  
      output.init_task()
  
      (
          origin_run_id,
          origin_step_name,
          origin_task_id,
      ) = clone_origin_task.split("/")
      # 2. initialize origin datastore
      origin = self.flow_datastore.get_task_datastore(
          origin_run_id, origin_step_name, origin_task_id
      )
      metadata_tags = ["attempt_id:{0}".format(retry_count)]
      output.clone(origin)
      self.metadata.register_metadata(
          run_id,
          step_name,
          task_id,
          [
              MetaDatum(
                  field="origin-task-id",
                  value=str(origin_task_id),
                  type="origin-task-id",
                  tags=metadata_tags,
              ),
              MetaDatum(
                  field="origin-run-id",
                  value=str(origin_run_id),
                  type="origin-run-id",
                  tags=metadata_tags,
              ),
              MetaDatum(
                  field="attempt",
                  value=str(retry_count),
                  type="attempt",
                  tags=metadata_tags,
              ),
          ],
      )
      output.done()
  
imports: from __future__ import print_function
  from io import BytesIO
  import math
  import sys
  import os
  import time
  from types import MethodType, FunctionType
  from metaflow.datastore.exceptions import DataException
  from .metaflow_config import MAX_ATTEMPTS
  from .metadata import MetaDatum
  from .mflog import TASK_LOG_SOURCE
  from .datastore import Inputs, TaskDataStoreSet
  from .exception import MetaflowInternalError, MetaflowDataMissing, MetaflowExceptionWrapper
  from .unbounded_foreach import UBF_CONTROL
  from .util import all_equal, get_username, resolve_identity, unicode_type
  from .current import current
  from metaflow.tracing import get_trace_id
  from collections import namedtuple
label_desc: Use image node_id 7 for calling the MetaflowTask obj's underlying member method code with example usage: obj.clone_only(step_name, run_id, task_id, clone_origin_task, retry_count, wait_only) without return types