This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 69859f8 [SPARK-38201][K8S] Fix `uploadFileToHadoopCompatibleFS` to
use `delSrc` and `overwrite` parameters
69859f8 is described below
commit 69859f81e5b13952b6e37fa4d51b1b4dbe19e5bc
Author: yangjie01 <[email protected]>
AuthorDate: Tue Feb 15 19:49:39 2022 -0800
[SPARK-38201][K8S] Fix `uploadFileToHadoopCompatibleFS` to use `delSrc` and
`overwrite` parameters
### What changes were proposed in this pull request?
`KubernetesUtils#uploadFileToHadoopCompatibleFS` defines the input
parameters `delSrc` and `overwrite`, but constants(`false` and `true`) are
used when invoke `FileSystem.copyFromLocalFile(boolean delSrc, boolean
overwrite, Path src, Path dst) `, this pr change to use passed in `delSrc` and
`overwrite` when invoke the`copyFromLocalFile` method.
### Why are the changes needed?
Bug fix
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Pass GA and add new test case
Closes #35509 from LuciferYang/SPARK-38201.
Lead-authored-by: yangjie01 <[email protected]>
Co-authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/deploy/k8s/KubernetesUtils.scala | 2 +-
.../spark/deploy/k8s/KubernetesUtilsSuite.scala | 66 +++++++++++++++++++++-
2 files changed, 65 insertions(+), 3 deletions(-)
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala
index 0c8d964..a05d07a 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala
@@ -344,7 +344,7 @@ object KubernetesUtils extends Logging {
delSrc : Boolean = false,
overwrite: Boolean = true): Unit = {
try {
- fs.copyFromLocalFile(false, true, src, dest)
+ fs.copyFromLocalFile(delSrc, overwrite, src, dest)
} catch {
case e: IOException =>
throw new SparkException(s"Error uploading file ${src.getName}", e)
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/KubernetesUtilsSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/KubernetesUtilsSuite.scala
index ef57a4b..5498238 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/KubernetesUtilsSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/KubernetesUtilsSuite.scala
@@ -17,13 +17,20 @@
package org.apache.spark.deploy.k8s
+import java.io.File
+import java.nio.charset.StandardCharsets
+
import scala.collection.JavaConverters._
import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder}
+import org.apache.commons.io.FileUtils
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.scalatest.PrivateMethodTester
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkException, SparkFunSuite}
-class KubernetesUtilsSuite extends SparkFunSuite {
+class KubernetesUtilsSuite extends SparkFunSuite with PrivateMethodTester {
private val HOST = "test-host"
private val POD = new PodBuilder()
.withNewSpec()
@@ -65,4 +72,59 @@ class KubernetesUtilsSuite extends SparkFunSuite {
assert(sparkPodWithNoContainerName.pod.getSpec.getHostname == HOST)
assert(sparkPodWithNoContainerName.container.getName == null)
}
+
+ test("SPARK-38201: check uploadFileToHadoopCompatibleFS with different
delSrc and overwrite") {
+ withTempDir { srcDir =>
+ withTempDir { destDir =>
+ val upload =
PrivateMethod[Unit](Symbol("uploadFileToHadoopCompatibleFS"))
+ val fileName = "test.txt"
+ val srcFile = new File(srcDir, fileName)
+ val src = new Path(srcFile.getAbsolutePath)
+ val dest = new Path(destDir.getAbsolutePath, fileName)
+ val fs = src.getFileSystem(new Configuration())
+
+ def checkUploadException(delSrc: Boolean, overwrite: Boolean): Unit = {
+ val message = intercept[SparkException] {
+ KubernetesUtils.invokePrivate(upload(src, dest, fs, delSrc,
overwrite))
+ }.getMessage
+ assert(message.contains("Error uploading file"))
+ }
+
+ def appendFileAndUpload(content: String, delSrc: Boolean, overwrite:
Boolean): Unit = {
+ FileUtils.write(srcFile, content, StandardCharsets.UTF_8, true)
+ KubernetesUtils.invokePrivate(upload(src, dest, fs, delSrc,
overwrite))
+ }
+
+ // Write a new file, upload file with delSrc = false and overwrite =
true.
+ // Upload successful and record the `fileLength`.
+ appendFileAndUpload("init-content", delSrc = false, overwrite = true)
+ val firstLength = fs.getFileStatus(dest).getLen
+
+ // Append the file, upload file with delSrc = false and overwrite =
true.
+ // Upload succeeded but `fileLength` changed.
+ appendFileAndUpload("append-content", delSrc = false, overwrite = true)
+ val secondLength = fs.getFileStatus(dest).getLen
+ assert(firstLength < secondLength)
+
+ // Upload file with delSrc = false and overwrite = false.
+ // Upload failed because dest exists and not changed.
+ checkUploadException(delSrc = false, overwrite = false)
+ assert(fs.exists(dest))
+ assert(fs.getFileStatus(dest).getLen == secondLength)
+
+ // Append the file again, upload file delSrc = true and overwrite =
true.
+ // Upload succeeded, `fileLength` changed and src not exists.
+ appendFileAndUpload("append-content", delSrc = true, overwrite = true)
+ val thirdLength = fs.getFileStatus(dest).getLen
+ assert(secondLength < thirdLength)
+ assert(!fs.exists(src))
+
+ // Rewrite a new file, upload file with delSrc = true and overwrite =
false.
+ // Upload failed because dest exists, src still exists.
+ FileUtils.write(srcFile, "re-init-content", StandardCharsets.UTF_8,
true)
+ checkUploadException(delSrc = true, overwrite = false)
+ assert(fs.exists(src))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
