Preserve natural line delimiters

Author: gerashegalov

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -19,7 +19,6 @@ package org.apache.spark.util
 
 import java.io._
 import java.lang.{Byte => JByte}
-import java.lang.InternalError
 import java.lang.management.{LockInfo, ManagementFactory, MonitorInfo, ThreadInfo}
 import java.lang.reflect.InvocationTargetException
 import java.math.{MathContext, RoundingMode}
@@ -2052,6 +2051,28 @@ private[spark] object Utils extends Logging {
     }
   }
 
+
+  private[this] val nonSpaceOrNaturalLineDelimiter: Char => Boolean = { ch => ch > ' ' || ch == '\r' || ch == '\n' }
+
+  /**
+   * Implements the same logic as JDK java.lang.String#trim by removing leading and trailing non-printable characters
+   * less or equal to '\u0020' (SPACE) but preserves natural line delimiters according to
+   * [[java.util.Properties]] load method. The natural line delimiters are removed by JDK during load.
+   * Therefore any remaining ones have been specifically provided and escaped by the user, and must not be ignored
+   *
+   * @param str
+   * @return the trimmed value of str
+   */
+  def trimExceptCRLF(str: String): String = {
+    val firstPos = str.indexWhere(nonSpaceOrNaturalLineDelimiter)
+    val lastPos = str.lastIndexWhere(nonSpaceOrNaturalLineDelimiter)
+    if (firstPos >= 0 && lastPos >= 0) {
+      str.substring(firstPos, lastPos + 1)
+    } else {
+      ""
+    }
+  }
+
   /** Load properties present in the given file. */
   def getPropertiesFromFile(filename: String): Map[String, String] = {
     val file = new File(filename)
@@ -2063,7 +2084,7 @@ private[spark] object Utils extends Logging {
       val properties = new Properties()
       properties.load(inReader)
       properties.stringPropertyNames().asScala
-        .map(k => (k, properties.getProperty(k)))
+        .map(k => (k, trimExceptCRLF(properties.getProperty(k))))
         .toMap
 
     } catch {

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -1145,31 +1145,44 @@ class SparkSubmitSuite
     conf1.get("spark.submit.pyFiles") should (startWith("/"))
   }
 
-  test("handles white space in --properties-file and --conf uniformly") {
-    val delimKey = "spark.my.delimiter"
-    val delimKeyFromFile = s"${delimKey}FromFile"
-    val newLine = "\n"
+  test("handles natural line delimiters in --properties-file and --conf uniformly") {
+    val delimKey = "spark.my.delimiter."
+    val LF = "\n"
+    val CR = "\r"
+
+    val leadingDelimKeyFromFile = s"${delimKey}leadingDelimKeyFromFile" -> s"${LF}blah"
+    val trailingDelimKeyFromFile = s"${delimKey}trailingDelimKeyFromFile" -> s"blah${CR}"
+    val infixDelimFromFile = s"${delimKey}infixDelimFromFile" -> s"${CR}blah${LF}"
+    val nonDelimSpaceFromFile = s"${delimKey}nonDelimSpaceFromFile" -> " blah\f"
+
+    val testProps = Seq(leadingDelimKeyFromFile, trailingDelimKeyFromFile, infixDelimFromFile, nonDelimSpaceFromFile)
+
     val props = new java.util.Properties()
     val propsFile = File.createTempFile("test-spark-conf", ".properties", Utils.createTempDir())
     val propsOutputStream = new FileOutputStream(propsFile)
     try {
-      props.put(delimKeyFromFile, newLine)
+      testProps.foreach { case (k, v) => props.put(k, v) }
       props.store(propsOutputStream, "test whitespace")
     } finally {
       propsOutputStream.close()
     }
 
     val clArgs = Seq(
       "--class", "org.SomeClass",
-      "--conf", s"${delimKey}=$newLine",
+      "--conf", s"${delimKey}=$LF",
       "--conf", "spark.master=yarn",
       "--properties-file", propsFile.getPath,
       "thejar.jar")
 
     val appArgs = new SparkSubmitArguments(clArgs)
     val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs)
-    conf.get(delimKey) should be (newLine)
-    conf.get(delimKeyFromFile) should be (newLine)
+
+    Seq((delimKey -> LF), leadingDelimKeyFromFile, trailingDelimKeyFromFile, infixDelimFromFile)
+      .foreach {
+        case (k, v) => conf.get(k) should be (v)
+      }
+
+    conf.get(nonDelimSpaceFromFile._1) should be ("blah")
   }
 }