code: def create_job(
      self,
      flow_name,
      run_id,
      step_name,
      task_id,
      attempt,
      user,
      code_package_sha,
      code_package_url,
      code_package_ds,
      step_cli,
      docker_image,
      docker_image_pull_policy,
      service_account=None,
      secrets=None,
      node_selector=None,
      namespace=None,
      cpu=None,
      gpu=None,
      gpu_vendor=None,
      disk=None,
      memory=None,
      use_tmpfs=None,
      tmpfs_tempdir=None,
      tmpfs_size=None,
      tmpfs_path=None,
      run_time_limit=None,
      env=None,
      persistent_volume_claims=None,
      tolerations=None,
      labels=None,
  ):
      if env is None:
          env = {}
  
      job = (
          KubernetesClient()
          .job(
              generate_name="t-{uid}-".format(uid=str(uuid4())[:8]),
              namespace=namespace,
              service_account=service_account,
              secrets=secrets,
              node_selector=node_selector,
              command=self._command(
                  flow_name=flow_name,
                  run_id=run_id,
                  step_name=step_name,
                  task_id=task_id,
                  attempt=attempt,
                  code_package_url=code_package_url,
                  step_cmds=[step_cli],
              ),
              image=docker_image,
              image_pull_policy=docker_image_pull_policy,
              cpu=cpu,
              memory=memory,
              disk=disk,
              gpu=gpu,
              gpu_vendor=gpu_vendor,
              timeout_in_seconds=run_time_limit,
              # Retries are handled by Metaflow runtime
              retries=0,
              step_name=step_name,
              tolerations=tolerations,
              labels=self._get_labels(labels),
              use_tmpfs=use_tmpfs,
              tmpfs_tempdir=tmpfs_tempdir,
              tmpfs_size=tmpfs_size,
              tmpfs_path=tmpfs_path,
              persistent_volume_claims=persistent_volume_claims,
          )
          .environment_variable("METAFLOW_CODE_SHA", code_package_sha)
          .environment_variable("METAFLOW_CODE_URL", code_package_url)
          .environment_variable("METAFLOW_CODE_DS", code_package_ds)
          .environment_variable("METAFLOW_USER", user)
          .environment_variable(
              "METAFLOW_SERVICE_URL", SERVICE_INTERNAL_URL
          )
          .environment_variable(
              "METAFLOW_SERVICE_HEADERS",
              json.dumps(SERVICE_HEADERS),
          )
          .environment_variable(
              "METAFLOW_DATASTORE_SYSROOT_S3", DATASTORE_SYSROOT_S3
          )
          .environment_variable(
              "METAFLOW_DATATOOLS_S3ROOT", DATATOOLS_S3ROOT
          )
          .environment_variable(
              "METAFLOW_DEFAULT_DATASTORE", self._datastore.TYPE
          )
          .environment_variable(
              "METAFLOW_DEFAULT_METADATA", DEFAULT_METADATA
          )
          .environment_variable("METAFLOW_KUBERNETES_WORKLOAD", 1)
          .environment_variable(
              "METAFLOW_KUBERNETES_FETCH_EC2_METADATA",
              KUBERNETES_FETCH_EC2_METADATA,
          )
          .environment_variable(
              "METAFLOW_RUNTIME_ENVIRONMENT", "kubernetes"
          )
          .environment_variable(
              "METAFLOW_DEFAULT_SECRETS_BACKEND_TYPE",
              DEFAULT_SECRETS_BACKEND_TYPE,
          )
          .environment_variable("METAFLOW_CARD_S3ROOT", CARD_S3ROOT)
          .environment_variable(
              "METAFLOW_DEFAULT_AWS_CLIENT_PROVIDER",
              DEFAULT_AWS_CLIENT_PROVIDER,
          )
          .environment_variable(
              "METAFLOW_AWS_SECRETS_MANAGER_DEFAULT_REGION",
              AWS_SECRETS_MANAGER_DEFAULT_REGION,
          )
          .environment_variable(
              "METAFLOW_S3_ENDPOINT_URL", S3_ENDPOINT_URL
          )
          .environment_variable(
              "METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT",
              AZURE_STORAGE_BLOB_SERVICE_ENDPOINT,
          )
          .environment_variable(
              "METAFLOW_DATASTORE_SYSROOT_AZURE",
              DATASTORE_SYSROOT_AZURE,
          )
          .environment_variable(
              "METAFLOW_CARD_AZUREROOT", CARD_AZUREROOT
          )
          .environment_variable(
              "METAFLOW_DATASTORE_SYSROOT_GS", DATASTORE_SYSROOT_GS
          )
          .environment_variable("METAFLOW_CARD_GSROOT", CARD_GSROOT)
          # support Metaflow sandboxes
          .environment_variable(
              "METAFLOW_INIT_SCRIPT", KUBERNETES_SANDBOX_INIT_SCRIPT
          )
          .environment_variable("METAFLOW_OTEL_ENDPOINT", OTEL_ENDPOINT)
          # Skip setting METAFLOW_DATASTORE_SYSROOT_LOCAL because metadata sync
          # between the local user instance and the remote Kubernetes pod
          # assumes metadata is stored in DATASTORE_LOCAL_DIR on the Kubernetes
          # pod; this happens when METAFLOW_DATASTORE_SYSROOT_LOCAL is NOT set (
          # see get_datastore_root_from_config in datastore/local.py).
      )
  
      # Temporary passing of *some* environment variables. Do not rely on this
      # mechanism as it will be removed in the near future
      for k, v in config_values():
          if k.startswith("METAFLOW_CONDA_") or k.startswith(
              "METAFLOW_DEBUG_"
          ):
              job.environment_variable(k, v)
  
      if S3_SERVER_SIDE_ENCRYPTION is not None:
          job.environment_variable(
              "METAFLOW_S3_SERVER_SIDE_ENCRYPTION",
              S3_SERVER_SIDE_ENCRYPTION,
          )
  
      # Set environment variables to support metaflow.integrations.ArgoEvent
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_WEBHOOK_URL",
          ARGO_EVENTS_INTERNAL_WEBHOOK_URL,
      )
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_EVENT", ARGO_EVENTS_EVENT
      )
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_EVENT_BUS", ARGO_EVENTS_EVENT_BUS
      )
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_EVENT_SOURCE", ARGO_EVENTS_EVENT_SOURCE
      )
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_SERVICE_ACCOUNT",
          ARGO_EVENTS_SERVICE_ACCOUNT,
      )
      job.environment_variable(
          "METAFLOW_ARGO_EVENTS_WEBHOOK_AUTH",
          ARGO_EVENTS_WEBHOOK_AUTH,
      )
  
      tmpfs_enabled = use_tmpfs or (tmpfs_size and not use_tmpfs)
      if tmpfs_enabled and tmpfs_tempdir:
          job.environment_variable("METAFLOW_TEMPDIR", tmpfs_path)
  
      for name, value in env.items():
          job.environment_variable(name, value)
  
      annotations = {
          "metaflow/user": user,
          "metaflow/flow_name": flow_name,
      }
      if current.get("project_name"):
          annotations.update(
              {
                  "metaflow/project_name": current.project_name,
                  "metaflow/branch_name": current.branch_name,
                  "metaflow/project_flow_name": current.project_flow_name,
              }
          )
  
      for name, value in annotations.items():
          job.annotation(name, value)
  
      (
          job.annotation("metaflow/run_id", run_id)
          .annotation("metaflow/step_name", step_name)
          .annotation("metaflow/task_id", task_id)
          .annotation("metaflow/attempt", attempt)
          .label("app.kubernetes.io/name", "metaflow-task")
          .label("app.kubernetes.io/part-of", "metaflow")
      )
  
      return job.create()
  
imports: import json
  import math
  import os
  import re
  import shlex
  import time
  from typing import Dict, List, Optional
  import uuid
  from uuid import uuid4
  from metaflow import current, util
  from metaflow.exception import MetaflowException
  from metaflow.metaflow_config import ARGO_EVENTS_EVENT, ARGO_EVENTS_EVENT_BUS, ARGO_EVENTS_EVENT_SOURCE, ARGO_EVENTS_SERVICE_ACCOUNT, ARGO_EVENTS_INTERNAL_WEBHOOK_URL, AWS_SECRETS_MANAGER_DEFAULT_REGION, ARGO_EVENTS_WEBHOOK_AUTH, AZURE_STORAGE_BLOB_SERVICE_ENDPOINT, CARD_AZUREROOT, CARD_GSROOT, CARD_S3ROOT, DATASTORE_SYSROOT_AZURE, DATASTORE_SYSROOT_GS, DATASTORE_SYSROOT_S3, DATATOOLS_S3ROOT, DEFAULT_AWS_CLIENT_PROVIDER, DEFAULT_METADATA, DEFAULT_SECRETS_BACKEND_TYPE, KUBERNETES_FETCH_EC2_METADATA, KUBERNETES_LABELS, KUBERNETES_SANDBOX_INIT_SCRIPT, S3_ENDPOINT_URL, SERVICE_HEADERS, SERVICE_INTERNAL_URL, S3_SERVER_SIDE_ENCRYPTION, OTEL_ENDPOINT
  from metaflow.metaflow_config_funcs import config_values
  from metaflow.mflog import BASH_SAVE_LOGS, bash_capture_logs, export_mflog_env_vars, get_log_tailer, tail_logs
  from .kubernetes_client import KubernetesClient
label_desc: Use image node_id 4 for calling the Kubernetes obj's underlying member method code with example usage: obj.create_job(flow_name, run_id, step_name, task_id, attempt, user, code_package_sha, code_package_url, code_package_ds, step_cli, docker_image, docker_image_pull_policy, service_account, secrets, node_selector, namespace, cpu, gpu, gpu_vendor, disk, memory, use_tmpfs, tmpfs_tempdir, tmpfs_size, tmpfs_path, run_time_limit, env, persistent_volume_claims, tolerations, labels) and returns: job