Resolve conflicts

Author: Robert Kruszewski

FORK.md

@@ -6,7 +6,6 @@
 * [SPARK-17059](https://issues.apache.org/jira/browse/SPARK-17059) - Allow FileFormat to specify partition pruning strategy via splits
 * [SPARK-24345](https://issues.apache.org/jira/browse/SPARK-24345) - Improve ParseError stop location when offending symbol is a token
 * [SPARK-23795](https://issues.apache.org/jira/browse/SPARK-23795) - Make AbstractLauncher#self() protected
-* [SPARK-23153](https://issues.apache.org/jira/browse/SPARK-23153) - Support application dependencies in submission client's local file system
 * [SPARK-18079](https://issues.apache.org/jira/browse/SPARK-18079) - CollectLimitExec.executeToIterator should perform per-partition limits
 
 * [SPARK-15777](https://issues.apache.org/jira/browse/SPARK-15777) (Partial fix) - Catalog federation

core/pom.xml

@@ -387,48 +387,17 @@
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
-
-    <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-crypto</artifactId>
-    </dependency>
-
-<<<<<<< HEAD
-    <!--
-     The following dependencies are depended upon in HiveCredentialProvider, but are only executed if Hive is enabled in
-     the user's Hadoop configuration.  So in order to prevent spark-core from depending on Hive, these deps have been
-     placed in the "provided" scope, rather than the "compile" scope, and NoClassDefFoundError exceptions are handled
-     when the user has not explicitly compiled with the Hive module.
-    -->
-    <dependency>
-      <groupId>${hive.group}</groupId>
-      <artifactId>hive-exec</artifactId>
-      <scope>provided</scope>
-    </dependency>
-    <dependency>
-      <groupId>${hive.group}</groupId>
-      <artifactId>hive-metastore</artifactId>
-      <scope>provided</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.thrift</groupId>
-      <artifactId>libthrift</artifactId>
-      <scope>provided</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.thrift</groupId>
-      <artifactId>libfb303</artifactId>
-      <scope>provided</scope>
-    </dependency>
-
     <dependency>
       <groupId>com.palantir.safe-logging</groupId>
       <artifactId>safe-logging</artifactId>
       <version>1.5.1</version>
     </dependency>
 
-=======
->>>>>>> master
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-crypto</artifactId>
+    </dependency>
+
   </dependencies>
   <build>
     <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -440,11 +440,7 @@ class SparkContext(config: SparkConf) extends SafeLogging {
     _statusTracker = new SparkStatusTracker(this, _statusStore)
 
     _progressBar =
-<<<<<<< HEAD
       if (_conf.get(UI_SHOW_CONSOLE_PROGRESS) && !safeLogIsInfoEnabled) {
-=======
-      if (_conf.get(UI_SHOW_CONSOLE_PROGRESS)) {
->>>>>>> master
         Some(new ConsoleProgressBar(this))
       } else {
         None
@@ -2488,35 +2484,18 @@ object SparkContext extends SafeLogging {
       Option(activeContext.get()).filter(_ ne sc).foreach { ctx =>
           val errMsg = "Only one SparkContext should be running in this JVM (see SPARK-2243)." +
             s"The currently running SparkContext was created at:\n${ctx.creationSite.longForm}"
-<<<<<<< HEAD
-          val exception = new SparkException(errMsg)
-          if (allowMultipleContexts) {
-            safeLogWarning("Multiple running SparkContexts detected in the same JVM!", exception)
-          } else {
-            throw exception
-          }
-=======
           throw new SparkException(errMsg)
->>>>>>> master
         }
 
       contextBeingConstructed.filter(_ ne sc).foreach { otherContext =>
         // Since otherContext might point to a partially-constructed context, guard against
         // its creationSite field being null:
         val otherContextCreationSite =
           Option(otherContext.creationSite).map(_.longForm).getOrElse("unknown location")
-<<<<<<< HEAD
         safeLogWarning("Another SparkContext is being constructed (or threw an exception in its" +
           " constructor).  This may indicate an error, since only one SparkContext may be" +
           " running in this JVM (see SPARK-2243).",
           UnsafeArg.of("otherContextCreationSite", otherContextCreationSite))
-=======
-        val warnMsg = "Another SparkContext is being constructed (or threw an exception in its" +
-          " constructor). This may indicate an error, since only one SparkContext should be" +
-          " running in this JVM (see SPARK-2243)." +
-          s" The other SparkContext was created at:\n$otherContextCreationSite"
-        logWarning(warnMsg)
->>>>>>> master
       }
     }
   }

core/src/main/scala/org/apache/spark/api/r/RRunner.scala

@@ -349,21 +349,17 @@ private[r] object RRunner {
     // "spark.sparkr.r.command" is deprecated and replaced by "spark.r.command",
     // but kept here for backward compatibility.
     val sparkConf = SparkEnv.get.conf
-<<<<<<< HEAD
-    val requestedRCommand = Provenance.fromConf("spark.r.command")
-      .orElse(Provenance.fromConf("spark.sparkr.r.command"))
+    val requestedRCommand = Provenance.fromConfOpt(sparkConf, R_COMMAND)
+      .getOrElse(Provenance.fromConf(sparkConf, SPARKR_COMMAND))
     val condaEnv = condaSetupInstructions.map(CondaEnvironmentManager.getOrCreateCondaEnvironment)
     val rCommand = condaEnv.map { conda =>
-      requestedRCommand.foreach(exec => sys.error(s"It's forbidden to set the r executable " +
-        s"when using conda, but found: $exec"))
+      if (requestedRCommand.value != SPARKR_COMMAND.defaultValue.get) {
+        sys.error(s"It's forbidden to set the r executable " +
+          s"when using conda, but found: ${requestedRCommand.value}")
+      }
 
       conda.condaEnvDir + "/bin/Rscript"
-    }.orElse(requestedRCommand.map(_.value))
-     .getOrElse("Rscript")
-=======
-    var rCommand = sparkConf.get(SPARKR_COMMAND)
-    rCommand = sparkConf.get(R_COMMAND).orElse(Some(rCommand)).get
->>>>>>> master
+    }.getOrElse(requestedRCommand.value)
 
     val rConnectionTimeout = sparkConf.get(R_BACKEND_CONNECTION_TIMEOUT)
     val rOptions = "--vanilla"

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

@@ -29,11 +29,8 @@ import scala.util.Random
 import com.palantir.logsafe.SafeArg
 
 import org.apache.spark._
-<<<<<<< HEAD
 import org.apache.spark.internal.SafeLogging
-=======
-import org.apache.spark.internal.{config, Logging}
->>>>>>> master
+import org.apache.spark.internal.config
 import org.apache.spark.io.CompressionCodec
 import org.apache.spark.serializer.Serializer
 import org.apache.spark.storage._
@@ -244,14 +241,11 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
               throw new SparkException(s"Failed to get locally stored broadcast data: $broadcastId")
             }
           case None =>
-<<<<<<< HEAD
-            safeLogInfo("Started reading broadcast variable",
-              SafeArg.of("id", id))
-=======
             val estimatedTotalSize = Utils.bytesToString(numBlocks * blockSize)
-            logInfo(s"Started reading broadcast variable $id with $numBlocks pieces " +
-              s"(estimated total size $estimatedTotalSize)")
->>>>>>> master
+            safeLogInfo("Started reading broadcast variable",
+              SafeArg.of("id", id),
+              SafeArg.of("numBlocks", numBlocks),
+              SafeArg.of("estimatedTotalSize", estimatedTotalSize))
             val startTimeMs = System.currentTimeMillis()
             val blocks = readBlocks()
             safeLogInfo("Reading broadcast variable finished",

core/src/main/scala/org/apache/spark/deploy/Common.scala

@@ -30,10 +30,14 @@ object Common {
       sys.props.get(sysPropName).map(Provenance(s"Spark config $sysPropName", _))
     }
 
-    def fromConf(sparkConf: SparkConf, conf: ConfigEntry[Option[String]]): Option[Provenance] = {
+    def fromConfOpt(sparkConf: SparkConf, conf: ConfigEntry[Option[String]]): Option[Provenance] = {
       sparkConf.get(conf).map(Provenance(s"Spark config ${conf.key}", _))
     }
 
+    def fromConf(sparkConf: SparkConf, conf: ConfigEntry[String]): Provenance = {
+      Provenance(s"Spark config ${conf.key}", sparkConf.get(conf))
+    }
+
     def fromConf(sparkConf: SparkConf, key: String): Option[Provenance] = {
       sparkConf.getOption(key).map(Provenance(s"Spark config $key", _))
     }

core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala

@@ -46,8 +46,8 @@ object PythonRunner extends CondaRunner with Logging {
     val otherArgs = args.slice(2, args.length)
     val sparkConf = new SparkConf()
     val secret = Utils.createSecret(sparkConf)
-    val presetPythonExec = Provenance.fromConf(sparkConf, PYSPARK_DRIVER_PYTHON)
-      .orElse(Provenance.fromConf(sparkConf, PYSPARK_PYTHON))
+    val presetPythonExec = Provenance.fromConfOpt(sparkConf, PYSPARK_DRIVER_PYTHON)
+      .orElse(Provenance.fromConfOpt(sparkConf, PYSPARK_PYTHON))
       .orElse(Provenance.fromEnv("PYSPARK_DRIVER_PYTHON"))
       .orElse(Provenance.fromEnv("PYSPARK_PYTHON"))
 

core/src/main/scala/org/apache/spark/deploy/RRunner.scala

@@ -25,20 +25,12 @@ import scala.collection.JavaConverters._
 
 import org.apache.hadoop.fs.Path
 
-<<<<<<< HEAD
-import org.apache.spark.SparkException
-import org.apache.spark.SparkUserAppException
+import org.apache.spark.{SparkException, SparkUserAppException}
 import org.apache.spark.api.conda.CondaEnvironment
-import org.apache.spark.api.r.RBackend
-import org.apache.spark.api.r.RUtils
-import org.apache.spark.api.r.SparkRDefaults
+import org.apache.spark.api.r.{RBackend, RUtils}
 import org.apache.spark.deploy.Common.Provenance
 import org.apache.spark.internal.Logging
-=======
-import org.apache.spark.{SparkException, SparkUserAppException}
-import org.apache.spark.api.r.{RBackend, RUtils}
 import org.apache.spark.internal.config.R._
->>>>>>> master
 import org.apache.spark.util.RedirectThread
 
 /**
@@ -62,27 +54,19 @@ object RRunner extends CondaRunner with Logging {
       }
       // "spark.sparkr.r.command" is deprecated and replaced by "spark.r.command",
       // but kept here for backward compatibility.
-<<<<<<< HEAD
       driverPreset
-        .orElse(Provenance.fromConf("spark.r.command"))
-        .orElse(Provenance.fromConf("spark.sparkr.r.command"))
-=======
-      var cmd = sys.props.getOrElse(SPARKR_COMMAND.key, SPARKR_COMMAND.defaultValue.get)
-      cmd = sys.props.getOrElse(R_COMMAND.key, cmd)
-      if (sys.props.getOrElse("spark.submit.deployMode", "client") == "client") {
-        cmd = sys.props.getOrElse("spark.r.driver.command", cmd)
-      }
-      cmd
->>>>>>> master
+        .orElse(Provenance.fromConf(R_COMMAND.key))
+        .orElse(Provenance.fromConf(SPARKR_COMMAND.key))
+        .getOrElse(Provenance(SPARKR_COMMAND.key, SPARKR_COMMAND.defaultValue.get))
     }
 
-    val rCommand: String = maybeConda.map { conda =>
-      presetRCommand.foreach { exec =>
-        sys.error(s"It's forbidden to configure the r command when using conda, but found: $exec")
+    val rCommand = maybeConda.map { conda =>
+      if (presetRCommand.value != SPARKR_COMMAND.defaultValueString) {
+        sys.error(s"It's forbidden to set the r executable " +
+          s"when using conda, but found: ${presetRCommand.value}")
       }
       conda.condaEnvDir + "/bin/Rscript"
-    }.orElse(presetRCommand.map(_.value))
-     .getOrElse("Rscript")
+    }.getOrElse(presetRCommand.value)
 
     //  Connection timeout set by R process on its connection to RBackend in seconds.
     val backendConnectionTimeout = sys.props.getOrElse(

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

@@ -98,16 +98,11 @@ private[spark] class MetricsSystem private (
   def start(registerStaticSources: Boolean = true) {
     require(!running, "Attempting to start a MetricsSystem that is already running")
     running = true
-<<<<<<< HEAD
-    StaticSources.allSources.foreach(registerSource)
-    registerSources()
-    SharedMetricRegistries.getDefault.addListener(defaultListener)
-=======
     if (registerStaticSources) {
       StaticSources.allSources.foreach(registerSource)
       registerSources()
     }
->>>>>>> master
+    SharedMetricRegistries.getDefault.addListener(defaultListener)
     registerSinks()
     sinks.foreach(_.start())
   }

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -21,7 +21,7 @@ import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
 import javax.annotation.concurrent.GuardedBy
 
-import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet}
+import scala.collection.mutable.{HashMap, HashSet}
 import scala.concurrent.Future
 
 import com.palantir.logsafe.SafeArg
@@ -31,13 +31,10 @@ import org.apache.hadoop.security.UserGroupInformation
 import org.apache.spark.{ExecutorAllocationClient, SparkEnv, SparkException, TaskState}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
-<<<<<<< HEAD
-import org.apache.spark.internal.SafeLogging
-=======
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.SafeLogging
 import org.apache.spark.internal.config._
 import org.apache.spark.internal.config.Network._
->>>>>>> master
 import org.apache.spark.rpc._
 import org.apache.spark.scheduler._
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
@@ -116,14 +113,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   private val reviveThread =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("driver-revive-thread")
 
-<<<<<<< HEAD
-  class DriverEndpoint(override val rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
-    extends ThreadSafeRpcEndpoint with SafeLogging {
-=======
   class DriverEndpoint extends ThreadSafeRpcEndpoint with Logging {
 
     override val rpcEnv: RpcEnv = CoarseGrainedSchedulerBackend.this.rpcEnv
->>>>>>> master
 
     // Executors that have been lost, but for which we don't yet know the real exit reason.
     protected val executorsPendingLossReason = new HashSet[String]