/** Deploy a Flink cluster with the given spec. */  
def deployCluster(id: Long, spec: FlinkDeploymentDef): IO[Throwable, TrackKey.ClusterKey] = {  
  for {  
    _       <- k8sCrOpr.applyDeployment(spec.copy(job = None))  
    trackKey = TrackKey.cluster(id, spec.namespace, spec.name)  
    _       <- obr.track(trackKey)  
  } yield trackKey  
}
// Submit FlinkDeployment CR to Kubernetes  
FlinkK8sOperator.deployCluster(deployRequest.id, flinkDeployDef).runIOAsTry match {  
  case Success(_) =>  
    logInfo(richMsg("Flink Cluster has been submitted successfully."))  
  case Failure(err) =>  
    logError(  
      richMsg(s"Submit Flink Cluster fail in${deployRequest.executionMode.getName}_V2 mode!"),  
      err)  
    throw err  
}

def runIOAsTry: Try[A] = unsafeRunToEither(io).toTry
/** unsafe run IO to Either. */  
@inline def unsafeRunToEither[E, A](zio: IO[E, A]): Either[Throwable, A] = Unsafe.unsafe {  
  implicit u =>  
    Runtime.default.unsafe  
      .run(zio.provideLayer(Runtime.removeDefaultLoggers >>> ZIOLogger.default))  
      .toEither  
}

/** Apply FlinkDeployment CR. */  
override def applyDeployment(spec: FlinkDeploymentDef): IO[Throwable, Unit] = {  
  lazy val mirrorSpace = s"${spec.namespace}_${spec.name}"  
  for {  
    _  <- ZIO.fail(FlinkDeploymentCRDNotFound()).unlessZIO(existFlinkDeploymentCRD)  
  } yield ()  
} *> ZIO.logInfo(s"Successfully apply FlinkDeployment K8s CR: namespace=${spec.namespace}, name=${spec.name}")