resolve conflicts

Author: Justin Uang

build/mvn

@@ -159,20 +159,12 @@ export MAVEN_OPTS=${MAVEN_OPTS:-"$_COMPILE_JVM_OPTS"}
 
 echo "Using \`mvn\` from path: $MVN_BIN" 1>&2
 
-<<<<<<< HEAD
-# Last, call the `mvn` command as usual
-"${MVN_BIN}" -DzincPort=${ZINC_PORT} -e "$@"
-||||||| merged common ancestors
-# Last, call the `mvn` command as usual
-"${MVN_BIN}" -DzincPort=${ZINC_PORT} "$@"
-=======
 # call the `mvn` command as usual
 # SPARK-25854
-"${MVN_BIN}" -DzincPort=${ZINC_PORT} "$@"
+"${MVN_BIN}" -DzincPort=${ZINC_PORT} -e "$@"
 MVN_RETCODE=$?
 
 # Try to shut down zinc explicitly if the server is still running.
 "${ZINC_BIN}" -shutdown -port ${ZINC_PORT}
 
 exit $MVN_RETCODE
->>>>>>> 3404a73~1

core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala

@@ -106,22 +106,12 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
     if (memoryMb.isDefined) {
       envVars.put("PYSPARK_EXECUTOR_MEMORY_MB", memoryMb.get.toString)
     }
-<<<<<<< HEAD
-    val worker: Socket = env.createPythonWorker(
-      pythonExec, envVars.asScala.toMap, condaInstructions)
-    // Whether is the worker released into idle pool
-    val released = new AtomicBoolean(false)
-||||||| merged common ancestors
-    val worker: Socket = env.createPythonWorker(pythonExec, envVars.asScala.toMap)
-    // Whether is the worker released into idle pool
-    val released = new AtomicBoolean(false)
-=======
-    val worker: Socket = env.createPythonWorker(pythonExec, envVars.asScala.toMap)
+    val worker: Socket = env.createPythonWorker(pythonExec, envVars.asScala.toMap,
+      condaInstructions)
     // Whether is the worker released into idle pool or closed. When any codes try to release or
     // close a worker, they should use `releasedOrClosed.compareAndSet` to flip the state to make
     // sure there is only one winner that is going to release or close the worker.
     val releasedOrClosed = new AtomicBoolean(false)
->>>>>>> 3404a73~1
 
     // Start a thread to feed the process input from our parent's iterator
     val writerThread = newWriterThread(env, worker, inputIterator, partitionIndex, context)
@@ -351,6 +341,7 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
           }
         }
         dataOut.flush()
+
         dataOut.writeInt(evalType)
         writeCommand(dataOut)
         writeIteratorToStream(dataOut)
@@ -476,18 +467,8 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       }
       // Check whether the worker is ready to be re-used.
       if (stream.readInt() == SpecialLengths.END_OF_STREAM) {
-<<<<<<< HEAD
-        if (reuseWorker) {
-          env.releasePythonWorker(pythonExec, envVars.asScala.toMap, condaInstructions, worker)
-          released.set(true)
-||||||| merged common ancestors
-        if (reuseWorker) {
-          env.releasePythonWorker(pythonExec, envVars.asScala.toMap, worker)
-          released.set(true)
-=======
         if (reuseWorker && releasedOrClosed.compareAndSet(false, true)) {
-          env.releasePythonWorker(pythonExec, envVars.asScala.toMap, worker)
->>>>>>> 3404a73~1
+          env.releasePythonWorker(pythonExec, envVars.asScala.toMap, condaInstructions, worker)
         }
       }
       eos = true

core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala

@@ -19,7 +19,6 @@ package org.apache.spark.api.python
 
 import java.io.{DataInputStream, DataOutputStream, EOFException, InputStream, OutputStreamWriter}
 import java.net.{InetAddress, ServerSocket, Socket, SocketException}
-import java.nio.charset.StandardCharsets
 import java.util.Arrays
 import javax.annotation.concurrent.GuardedBy
 
@@ -33,18 +32,10 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.security.SocketAuthHelper
 import org.apache.spark.util.{RedirectThread, Utils}
 
-<<<<<<< HEAD
 private[spark] class PythonWorkerFactory(requestedPythonExec: Option[String],
                                          requestedEnvVars: Map[String, String],
                                          condaInstructions: Option[CondaSetupInstructions])
-  extends Logging {
-||||||| merged common ancestors
-private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String, String])
-  extends Logging {
-=======
-private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String, String])
   extends Logging { self =>
->>>>>>> 3404a73~1
 
   import PythonWorkerFactory._
 
@@ -101,7 +92,6 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
   @GuardedBy("self")
   private val simpleWorkers = new mutable.WeakHashMap[Socket, Process]()
 
-<<<<<<< HEAD
   private[this] val condaEnv = {
     // Set up conda environment if there are any conda packages requested
     condaInstructions.map(CondaEnvironmentManager.createCondaEnvironment)
@@ -122,11 +112,6 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
   }
 
   val pythonPath = PythonUtils.mergePythonPaths(
-||||||| merged common ancestors
-  val pythonPath = PythonUtils.mergePythonPaths(
-=======
-  private val pythonPath = PythonUtils.mergePythonPaths(
->>>>>>> 3404a73~1
     PythonUtils.sparkPythonPath,
     envVars.getOrElse("PYTHONPATH", ""),
     sys.env.getOrElse("PYTHONPATH", ""))