code: def _new_task(
      self,
      run_id,
      step_name,
      task_id=None,
      attempt=0,
      tags=None,
      sys_tags=None,
  ):
      # first ensure that the step exists
      self._get_or_create("step", run_id, step_name)
      task, did_create = self._get_or_create(
          "task",
          run_id,
          step_name,
          task_id,
          tags=tags,
          sys_tags=sys_tags,
      )
      if did_create:
          self._register_system_metadata(
              run_id, step_name, task["task_id"], attempt
          )
      return task["task_id"], did_create
  
imports: import os
  import random
  import requests
  import time
  from metaflow.exception import MetaflowException, MetaflowTaggingError, MetaflowInternalError
  from metaflow.metaflow_config import SERVICE_RETRY_COUNT, SERVICE_HEADERS, SERVICE_URL
  from metaflow.metadata import MetadataProvider
  from metaflow.metadata.heartbeat import HB_URL_KEY
  from metaflow.sidecar import Message, MessageTypes, Sidecar
  from metaflow.util import version_parse
label_desc: Use image node_id 19 for calling the ServiceMetadataProvider obj's underlying member method code with example usage: obj._new_task(run_id, step_name, task_id, attempt, tags, sys_tags) and returns: task,  did_create