[SPARK-40817][K8S][3.2] spark.files should preserve remote files

### What changes were proposed in this pull request?
Backport apache#38376 to `branch-3.2`

You can find a detailed description of the issue and an example reproduction on the Jira card: https://issues.apache.org/jira/browse/SPARK-40817

The idea for this fix is to update the logic which uploads user-specified files (via `spark.jars`, `spark.files`, etc) to `spark.kubernetes.file.upload.path`. After uploading local files, it used to overwrite the initial list of URIs passed by the user and it would thus erase all remote URIs which were specified there.

Small example of this behaviour:
1. User set the value of `spark.jars` to `s3a://some-bucket/my-application.jar,/tmp/some-local-jar.jar` when running `spark-submit` in cluster mode
2. `BasicDriverFeatureStep.getAdditionalPodSystemProperties()` gets called at one point while running `spark-submit`
3. This function would set `spark.jars` to a new value of `${SPARK_KUBERNETES_UPLOAD_PATH}/spark-upload-${RANDOM_STRING}/some-local-jar.jar`. Note that `s3a://some-bucket/my-application.jar` has been discarded.

With the logic proposed in this PR, the new value of `spark.jars` would be `s3a://some-bucket/my-application.jar,${SPARK_KUBERNETES_UPLOAD_PATH}/spark-upload-${RANDOM_STRING}/some-local-jar.jar`, so in other words we are making sure that remote URIs are no longer discarded.

### Why are the changes needed?
We encountered this issue in production when trying to launch Spark on Kubernetes jobs in cluster mode with a fix of local and remote dependencies.

### Does this PR introduce _any_ user-facing change?
Yes, see description of the new behaviour above.

### How was this patch tested?
- Added a unit test for the new behaviour
- Added an integration test for the new behaviour
- Tried this patch in our Kubernetes environment with `SparkPi`:
```
spark-submit \
  --master k8s://https://$KUBERNETES_API_SERVER_URL:443 \
  --deploy-mode cluster \
  --name=spark-submit-test \
  --class org.apache.spark.examples.SparkPi \
  --conf spark.jars=/opt/my-local-jar.jar,s3a://$BUCKET_NAME/my-remote-jar.jar \
  --conf spark.kubernetes.file.upload.path=s3a://$BUCKET_NAME/my-upload-path/ \
  [...]
  /opt/spark/examples/jars/spark-examples_2.12-3.1.3.jar
```
Before applying the patch, `s3a://$BUCKET_NAME/my-remote-jar.jar` was discarded from the final value of `spark.jars`. After applying the patch and launching the job again, I confirmed that `s3a://$BUCKET_NAME/my-remote-jar.jar` was no longer discarded by looking at the Spark config for the running job.

Closes apache#39670 from antonipp/spark-40817-branch-3.2.

Authored-by: Anton Ippolitov <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: antonipp

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStep.scala

@@ -162,27 +162,27 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
       MEMORY_OVERHEAD_FACTOR.key -> overheadFactor.toString)
     // try upload local, resolvable files to a hadoop compatible file system
     Seq(JARS, FILES, ARCHIVES, SUBMIT_PYTHON_FILES).foreach { key =>
-      val uris = conf.get(key).filter(uri => KubernetesUtils.isLocalAndResolvable(uri))
+      val (localUris, remoteUris) =
+        conf.get(key).partition(uri => KubernetesUtils.isLocalAndResolvable(uri))
       val value = {
         if (key == ARCHIVES) {
-          uris.map(UriBuilder.fromUri(_).fragment(null).build()).map(_.toString)
+          localUris.map(UriBuilder.fromUri(_).fragment(null).build()).map(_.toString)
         } else {
-          uris
+          localUris
         }
       }
       val resolved = KubernetesUtils.uploadAndTransformFileUris(value, Some(conf.sparkConf))
       if (resolved.nonEmpty) {
         val resolvedValue = if (key == ARCHIVES) {
-          uris.zip(resolved).map { case (uri, r) =>
+          localUris.zip(resolved).map { case (uri, r) =>
             UriBuilder.fromUri(r).fragment(new java.net.URI(uri).getFragment).build().toString
           }
         } else {
           resolved
         }
-        additionalProps.put(key.key, resolvedValue.mkString(","))
+        additionalProps.put(key.key, (resolvedValue ++ remoteUris).mkString(","))
       }
     }
     additionalProps.toMap
   }
 }
-

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.deploy.k8s.features
 import scala.collection.JavaConverters._
 
 import io.fabric8.kubernetes.api.model.{ContainerPort, ContainerPortBuilder, LocalObjectReferenceBuilder, Quantity}
+import org.apache.hadoop.fs.{LocalFileSystem, Path}
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s.{KubernetesTestConf, SparkPod}
@@ -232,6 +233,33 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
     assert(portMap2(BLOCK_MANAGER_PORT_NAME) === 1235)
   }
 
+  test("SPARK-40817: Check that remote JARs do not get discarded in spark.jars") {
+    val FILE_UPLOAD_PATH = "s3a://some-bucket/upload-path"
+    val REMOTE_JAR_URI = "s3a://some-bucket/my-application.jar"
+    val LOCAL_JAR_URI = "/tmp/some-local-jar.jar"
+
+    val sparkConf = new SparkConf()
+      .set(CONTAINER_IMAGE, "spark-driver:latest")
+      .set(JARS, Seq(REMOTE_JAR_URI, LOCAL_JAR_URI))
+      .set(KUBERNETES_FILE_UPLOAD_PATH, FILE_UPLOAD_PATH)
+      // Instead of using the real S3A Hadoop driver, use a fake local one
+      .set("spark.hadoop.fs.s3a.impl", classOf[TestFileSystem].getCanonicalName)
+      .set("spark.hadoop.fs.s3a.impl.disable.cache", "true")
+    val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val featureStep = new BasicDriverFeatureStep(kubernetesConf)
+
+    val sparkJars = featureStep.getAdditionalPodSystemProperties()(JARS.key).split(",")
+
+    // Both the remote and the local JAR should be there
+    assert(sparkJars.size == 2)
+    // The remote JAR path should have been left untouched
+    assert(sparkJars.contains(REMOTE_JAR_URI))
+    // The local JAR should have been uploaded to spark.kubernetes.file.upload.path
+    assert(!sparkJars.contains(LOCAL_JAR_URI))
+    assert(sparkJars.exists(path =>
+      path.startsWith(FILE_UPLOAD_PATH) && path.endsWith("some-local-jar.jar")))
+  }
+
   def containerPort(name: String, portNumber: Int): ContainerPort =
     new ContainerPortBuilder()
       .withName(name)
@@ -241,3 +269,16 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
 
   private def amountAndFormat(quantity: Quantity): String = quantity.getAmount + quantity.getFormat
 }
+
+/**
+ * No-op Hadoop FileSystem
+ */
+private class TestFileSystem extends LocalFileSystem {
+  override def copyFromLocalFile(
+    delSrc: Boolean,
+    overwrite: Boolean,
+    src: Path,
+    dst: Path): Unit = {}
+
+  override def mkdirs(path: Path): Boolean = true
+}

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala

@@ -32,7 +32,7 @@ import org.scalatest.time.{Minutes, Span}
 
 import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.integrationtest.DepsTestsSuite.{DEPS_TIMEOUT, FILE_CONTENTS, HOST_PATH}
-import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite.{INTERVAL, MinikubeTag, TIMEOUT}
+import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite.{INTERVAL, MinikubeTag, SPARK_PI_MAIN_CLASS, TIMEOUT}
 import org.apache.spark.deploy.k8s.integrationtest.Utils.getExamplesJarName
 import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.Minikube
 import org.apache.spark.internal.config.{ARCHIVES, PYSPARK_DRIVER_PYTHON, PYSPARK_PYTHON}
@@ -167,6 +167,42 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     })
   }
 
+  test(
+    "SPARK-40817: Check that remote files do not get discarded in spark.files",
+    k8sTestTag,
+    MinikubeTag) {
+    tryDepsTest({
+      // Create a local file
+      val localFileName = Utils.createTempFile(FILE_CONTENTS, HOST_PATH)
+
+      // Create a remote file on S3
+      val remoteFileName = "some-remote-file.txt"
+      val remoteFileKey = s"some-path/${remoteFileName}"
+      createS3Object(remoteFileKey, "Some Content")
+      val remoteFileFullPath = s"s3a://${BUCKET}/${remoteFileKey}"
+
+      // Put both file paths in spark.files
+      sparkAppConf.set("spark.files", s"$HOST_PATH/$localFileName,${remoteFileFullPath}")
+      // Allows to properly read executor logs once the job is finished
+      sparkAppConf.set("spark.kubernetes.executor.deleteOnTermination", "false")
+
+      // Run SparkPi and make sure that both files have been properly downloaded on running pods
+      val examplesJar = Utils.getTestFileAbsolutePath(getExamplesJarName(), sparkHomeDir)
+      runSparkApplicationAndVerifyCompletion(
+        appResource = examplesJar,
+        mainClass = SPARK_PI_MAIN_CLASS,
+        appArgs = Array(),
+        expectedDriverLogOnCompletion = Seq("Pi is roughly 3"),
+        // We can check whether the Executor pod has successfully
+        // downloaded both the local and the remote file
+        expectedExecutorLogOnCompletion = Seq(localFileName, remoteFileName),
+        driverPodChecker = doBasicDriverPodCheck,
+        executorPodChecker = doBasicExecutorPodCheck,
+        isJVM = true
+      )
+    })
+  }
+
   test("SPARK-33615: Launcher client archives", k8sTestTag, MinikubeTag) {
     tryDepsTest {
       val fileName = Utils.createTempFile(FILE_CONTENTS, HOST_PATH)
@@ -259,12 +295,20 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     }
   }
 
+  private def getS3Client(
+      endPoint: String,
+      accessKey: String = ACCESS_KEY,
+      secretKey: String = SECRET_KEY): AmazonS3Client = {
+    val credentials = new BasicAWSCredentials(accessKey, secretKey)
+    val s3client = new AmazonS3Client(credentials)
+    s3client.setEndpoint(endPoint)
+    s3client
+  }
+
   private def createS3Bucket(accessKey: String, secretKey: String, endPoint: String): Unit = {
     Eventually.eventually(TIMEOUT, INTERVAL) {
       try {
-        val credentials = new BasicAWSCredentials(accessKey, secretKey)
-        val s3client = new AmazonS3Client(credentials)
-        s3client.setEndpoint(endPoint)
+        val s3client = getS3Client(endPoint, accessKey, secretKey)
         s3client.createBucket(BUCKET)
       } catch {
         case e: Exception =>
@@ -273,6 +317,21 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     }
   }
 
+  private def createS3Object(
+      objectKey: String,
+      objectContent: String,
+      endPoint: String = getServiceUrl(svcName)): Unit = {
+    Eventually.eventually(TIMEOUT, INTERVAL) {
+      try {
+        val s3client = getS3Client(endPoint)
+        s3client.putObject(BUCKET, objectKey, objectContent)
+      } catch {
+        case e: Exception =>
+          throw new SparkException(s"Failed to create object $BUCKET/$objectKey.", e)
+      }
+    }
+  }
+
   private def getServiceUrl(serviceName: String): String = {
     val fuzzyUrlMatcher = """^(.*?)([a-zA-Z]+://.*?)(\s*)$""".r
     Eventually.eventually(TIMEOUT, INTERVAL) {