diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 433c45d51fd6b..b9b275c190fee 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -84,8 +84,10 @@ private[spark] class Client( def run(): Unit = { logInfo(s"Starting application $kubernetesAppId in Kubernetes...") - val (driverSubmitSslOptions, isKeyStoreLocalFile) = parseDriverSubmitSslOptions() + Seq(uploadedFiles, uploadedJars, Some(mainAppResource)).foreach(checkForFilesExistence) + + val (driverSubmitSslOptions, isKeyStoreLocalFile) = parseDriverSubmitSslOptions() val parsedCustomLabels = parseCustomLabels(customLabels) var k8ConfBuilder = new K8SConfigBuilder() .withApiVersion("v1") @@ -661,6 +663,22 @@ private[spark] class Client( }).toMap }).getOrElse(Map.empty[String, String]) } + + private def checkForFilesExistence(maybePaths: Option[String]): Unit = { + maybePaths.foreach { paths => + paths.split(",").foreach { path => + val uri = Utils.resolveURI(path) + uri.getScheme match { + case "file" | null => + val file = new File(uri.getPath) + if (!file.isFile) { + throw new SparkException(s"""file "${uri}" does not exist!""") + } + case _ => + } + } + } + } } private[spark] object Client extends Logging {