Skip to content

Commit edcb878

Browse files
zuotingbingMarcelo Vanzin
authored andcommitted
[SPARK-20338][CORE] Spaces in spark.eventLog.dir are not correctly handled
## What changes were proposed in this pull request? “spark.eventLog.dir” supports with space characters. 1. Update EventLoggingListenerSuite like `testDir = Utils.createTempDir(namePrefix = s"history log")` 2. Fix EventLoggingListenerSuite tests ## How was this patch tested? update unit tests Author: zuotingbing <zuo.tingbing9@zte.com.cn> Closes #18285 from zuotingbing/spark-resolveURI.
1 parent 53e48f7 commit edcb878

File tree

3 files changed

+7
-9
lines changed

3 files changed

+7
-9
lines changed

core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -96,8 +96,8 @@ private[spark] class EventLoggingListener(
9696
}
9797

9898
val workingPath = logPath + IN_PROGRESS
99-
val uri = new URI(workingPath)
10099
val path = new Path(workingPath)
100+
val uri = path.toUri
101101
val defaultFs = FileSystem.getDefaultUri(hadoopConf).getScheme
102102
val isDefaultLocal = defaultFs == null || defaultFs == "file"
103103

@@ -320,7 +320,7 @@ private[spark] object EventLoggingListener extends Logging {
320320
appId: String,
321321
appAttemptId: Option[String],
322322
compressionCodecName: Option[String] = None): String = {
323-
val base = logBaseDir.toString.stripSuffix("/") + "/" + sanitize(appId)
323+
val base = new Path(logBaseDir).toString.stripSuffix("/") + "/" + sanitize(appId)
324324
val codec = compressionCodecName.map("." + _).getOrElse("")
325325
if (appAttemptId.isDefined) {
326326
base + "_" + sanitize(appAttemptId.get) + codec

core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package org.apache.spark.deploy.history
1919

2020
import java.io._
21-
import java.net.URI
2221
import java.nio.charset.StandardCharsets
2322
import java.util.concurrent.TimeUnit
2423
import java.util.zip.{ZipInputStream, ZipOutputStream}
@@ -27,7 +26,7 @@ import scala.concurrent.duration._
2726
import scala.language.postfixOps
2827

2928
import com.google.common.io.{ByteStreams, Files}
30-
import org.apache.hadoop.fs.FileStatus
29+
import org.apache.hadoop.fs.{FileStatus, Path}
3130
import org.apache.hadoop.hdfs.DistributedFileSystem
3231
import org.json4s.jackson.JsonMethods._
3332
import org.mockito.Matchers.any
@@ -63,7 +62,7 @@ class FsHistoryProviderSuite extends SparkFunSuite with BeforeAndAfter with Matc
6362
codec: Option[String] = None): File = {
6463
val ip = if (inProgress) EventLoggingListener.IN_PROGRESS else ""
6564
val logUri = EventLoggingListener.getLogPath(testDir.toURI, appId, appAttemptId)
66-
val logPath = new URI(logUri).getPath + ip
65+
val logPath = new Path(logUri).toUri.getPath + ip
6766
new File(logPath)
6867
}
6968

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package org.apache.spark.scheduler
1919

2020
import java.io.{File, FileOutputStream, InputStream, IOException}
21-
import java.net.URI
2221

2322
import scala.collection.mutable
2423
import scala.io.Source
@@ -52,7 +51,7 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
5251
private var testDirPath: Path = _
5352

5453
before {
55-
testDir = Utils.createTempDir()
54+
testDir = Utils.createTempDir(namePrefix = s"history log")
5655
testDir.deleteOnExit()
5756
testDirPath = new Path(testDir.getAbsolutePath())
5857
}
@@ -111,7 +110,7 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
111110

112111
test("Log overwriting") {
113112
val logUri = EventLoggingListener.getLogPath(testDir.toURI, "test", None)
114-
val logPath = new URI(logUri).getPath
113+
val logPath = new Path(logUri).toUri.getPath
115114
// Create file before writing the event log
116115
new FileOutputStream(new File(logPath)).close()
117116
// Expected IOException, since we haven't enabled log overwrite.
@@ -293,7 +292,7 @@ object EventLoggingListenerSuite {
293292
val conf = new SparkConf
294293
conf.set("spark.eventLog.enabled", "true")
295294
conf.set("spark.eventLog.testing", "true")
296-
conf.set("spark.eventLog.dir", logDir.toUri.toString)
295+
conf.set("spark.eventLog.dir", logDir.toString)
297296
compressionCodec.foreach { codec =>
298297
conf.set("spark.eventLog.compress", "true")
299298
conf.set("spark.io.compression.codec", codec)

0 commit comments

Comments
 (0)