[LIVY-498][REPL] Fix Windows CRLF line ending issue in SparkR interpreter

## What changes were proposed in this pull request?

If the issued query contains CRLF EOL, it will be failed to execute on *nix machine. This happens when submitting queries from Windows machine and executing on Linux machine.

So here propose to convert statement to match system's EOL.

## How was this patch tested?

New UT added.

Author: jerryshao <sshao@hortonworks.com>

Closes #105 from jerryshao/LIVY-498.

core/src/main/scala/org/apache/livy/EOLUtils.scala

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy
+
+/**
+ * Helper class to deal with end-of-line markers in text files.
+ */
+object EOLUtils {
+  /** Unix-style end-of-line marker (LF) */
+  private val EOL_UNIX: String = "\n"
+
+  /** Windows-style end-of-line marker (CRLF) */
+  private val EOL_WINDOWS: String = "\r\n"
+
+  /** "Old Mac"-style end-of-line marker (CR) */
+  private val EOL_OLD_MAC: String = "\r"
+
+  /** Default end-of-line marker on current syste */
+  private val EOL_SYSTEM_DEFAULT: String = System.getProperty("line.separator")
+
+  object Mode extends Enumeration {
+    type Mode = Value
+
+    val LF, CRLF, CR = Value
+
+    lazy val SYSTEM_DEFAULT: Mode = {
+      val tmp = if (EOL_SYSTEM_DEFAULT == EOL_UNIX) {
+        LF
+      } else if (EOL_SYSTEM_DEFAULT == EOL_WINDOWS) {
+        CRLF
+      } else if (EOL_SYSTEM_DEFAULT == EOL_OLD_MAC) {
+        CR
+      } else {
+        null
+      }
+
+      if (tmp == null) {
+        throw new IllegalStateException("Could not determine system default end-of-line marker")
+      }
+      tmp
+    }
+
+    private def determineEOL(s: String): Mode = {
+      val charArray = s.toCharArray
+
+      var prev: Char = null.asInstanceOf[Char]
+      for (ch <- charArray) {
+        if (ch == '\n') {
+          if (prev == '\r') {
+            return CRLF
+          } else {
+            return LF
+          }
+        } else if (prev == '\r') {
+          return CR
+        }
+
+        prev = ch
+      }
+
+      null
+    }
+
+    def hasWindowsEOL(s: String): Boolean = determineEOL(s) == CRLF
+
+    def hasUnixEOL(s: String): Boolean = determineEOL(s) == LF
+
+    def hasOldMacEOL(s: String): Boolean = determineEOL(s) == CR
+
+    def hasSystemDefaultEOL(s: String): Boolean = determineEOL(s) == SYSTEM_DEFAULT
+  }
+
+  def convertToSystemEOL(s: String): String = convertLineEndings(s, EOL_SYSTEM_DEFAULT)
+
+  private def convertLineEndings(s: String, eol: String): String = {
+    if (Mode.hasWindowsEOL(s)) {
+      s.replaceAll(EOL_WINDOWS, eol)
+    } else if (Mode.hasUnixEOL(s)) {
+      s.replaceAll(EOL_UNIX, eol)
+    } else if (Mode.hasOldMacEOL(s)) {
+      s.replaceAll(EOL_OLD_MAC, eol)
+    } else {
+      s
+    }
+  }
+}

core/src/test/scala/org/apache/livy/EOLUtilsSuite.scala

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy
+
+import org.scalatest.FunSuite
+
+class EOLUtilsSuite extends FunSuite with LivyBaseUnitTestSuite {
+
+  test("check EOL") {
+    val s1 = "test\r\ntest"
+    assert(!EOLUtils.Mode.hasUnixEOL(s1))
+    assert(!EOLUtils.Mode.hasOldMacEOL(s1))
+    assert(EOLUtils.Mode.hasWindowsEOL(s1))
+
+    val s2 = "test\ntest"
+    assert(EOLUtils.Mode.hasUnixEOL(s2))
+    assert(!EOLUtils.Mode.hasOldMacEOL(s2))
+    assert(!EOLUtils.Mode.hasWindowsEOL(s2))
+
+    val s3 = "test\rtest"
+    assert(!EOLUtils.Mode.hasUnixEOL(s3))
+    assert(EOLUtils.Mode.hasOldMacEOL(s3))
+    assert(!EOLUtils.Mode.hasWindowsEOL(s3))
+
+    val s4 = "testtest"
+    assert(!EOLUtils.Mode.hasUnixEOL(s4))
+    assert(!EOLUtils.Mode.hasOldMacEOL(s4))
+    assert(!EOLUtils.Mode.hasWindowsEOL(s4))
+  }
+
+  test("convert EOL") {
+    val s1 = "test\r\ntest"
+    val s2 = "test\ntest"
+    val s3 = "test\rtest"
+    val s4 = "testtest"
+
+    assert(EOLUtils.convertToSystemEOL(s1) === EOLUtils.convertToSystemEOL(s2))
+    assert(EOLUtils.convertToSystemEOL(s1) === EOLUtils.convertToSystemEOL(s3))
+    assert(EOLUtils.convertToSystemEOL(s2) === EOLUtils.convertToSystemEOL(s3))
+    assert(EOLUtils.convertToSystemEOL(s4) === s4)
+  }
+}

integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala

@@ -97,6 +97,7 @@ class InteractiveIT extends BaseIntegrationTestSuite {
       s.run("%table x").verifyResult(".*headers.*type.*name.*data.*")
       s.run("abcde").verifyError(ename = "NameError", evalue = "name 'abcde' is not defined")
       s.run("raise KeyError, 'foo'").verifyError(ename = "KeyError", evalue = "'foo'")
+      s.run("print(1)\r\nprint(1)").verifyResult("1\n1")
     }
   }
 
@@ -115,6 +116,7 @@ class InteractiveIT extends BaseIntegrationTestSuite {
         """|root
           | |-- name: string (nullable = true)
           | |-- age: double (nullable = true)""".stripMargin))
+      s.run("print(1)\r\nprint(1)").verifyResult(".*1\n.*1")
     }
   }
 

repl/src/main/scala/org/apache/livy/repl/ReplDriver.scala

@@ -23,7 +23,7 @@ import scala.concurrent.duration.Duration
 import io.netty.channel.ChannelHandlerContext
 import org.apache.spark.SparkConf
 
-import org.apache.livy.Logging
+import org.apache.livy.{EOLUtils, Logging}
 import org.apache.livy.client.common.ClientConf
 import org.apache.livy.rsc.{BaseProtocol, ReplJobResults, RSCConf}
 import org.apache.livy.rsc.BaseProtocol.ReplState
@@ -55,7 +55,7 @@ class ReplDriver(conf: SparkConf, livyConf: RSCConf)
   }
 
   def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.ReplJobRequest): Int = {
-    session.execute(msg.code, msg.codeType)
+    session.execute(EOLUtils.convertToSystemEOL(msg.code), msg.codeType)
   }
 
   def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.CancelReplJobRequest): Unit = {
@@ -63,7 +63,7 @@ class ReplDriver(conf: SparkConf, livyConf: RSCConf)
   }
 
   def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.ReplCompleteRequest): Array[String] = {
-    session.complete(msg.code, msg.codeType, msg.cursor)
+    session.complete(EOLUtils.convertToSystemEOL(msg.code), msg.codeType, msg.cursor)
   }
 
   /**