Update based on review

Author: mnazbro

core/src/main/scala/org/apache/spark/api/conda/CondaEnvironmentManager.scala

@@ -20,7 +20,7 @@ import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
 import java.nio.file.attribute.PosixFilePermission
-import java.util.concurrent.locks.{Lock, ReentrantLock}
+import java.util.concurrent.locks.ReentrantLock
 import java.util.regex.Pattern
 
 import scala.collection.JavaConverters._
@@ -213,8 +213,9 @@ object CondaEnvironmentManager extends Logging {
   private[this] val httpUrlToken =
     Pattern.compile("(\\b\\w+://[^:/@]*:)([^/@]+)(?=@([\\w-.]+)(:\\d+)?\\b)")
 
-  private val initializedEnvironmentsLock = new ReentrantLock()
-  private val initializedEnvironments = mutable.HashMap[CondaSetupInstructions, CondaEnvironment]()
+  private[this] val initializedEnvironmentsLock = new ReentrantLock()
+  private[this] val initializedEnvironments =
+    mutable.HashMap[CondaSetupInstructions, CondaEnvironment]()
 
   private[conda] def redactCredentials(line: String): String = {
     httpUrlToken.matcher(line).replaceAll("$1<password>")
@@ -228,11 +229,8 @@ object CondaEnvironmentManager extends Logging {
     new CondaEnvironmentManager(condaBinaryPath, verbosity, packageDirs)
   }
 
-  /**
-   * Helper method to create a conda environment from [[CondaEnvironment.CondaSetupInstructions]].
-   * This is intended to be called on the executor with serialized instructions.
-   */
-  def createCondaEnvironment(instructions: CondaSetupInstructions): CondaEnvironment = {
+  private[this] def createCondaEnvironment(
+            instructions: CondaSetupInstructions): CondaEnvironment = {
     val condaPackages = instructions.packages
     val env = SparkEnv.get
     val condaEnvManager = CondaEnvironmentManager.fromConf(env.conf)
@@ -247,9 +245,10 @@ object CondaEnvironmentManager extends Logging {
   }
 
   /**
-   * Helper method that will cache precreated conda environments.
+   * Gets the conda environment for [[CondaEnvironment.CondaSetupInstructions]]. This will create
+   * a new one if none exists.
    */
-  def createOrGetCondaEnvironment(instructions: CondaSetupInstructions): CondaEnvironment = {
+  def getOrCreateCondaEnvironment(instructions: CondaSetupInstructions): CondaEnvironment = {
     initializedEnvironmentsLock.lock()
     try {
       initializedEnvironments.get(instructions) match {

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

@@ -94,7 +94,7 @@ private[spark] class PythonWorkerFactory(requestedPythonExec: Option[String],
 
   private[this] val condaEnv = {
     // Set up conda environment if there are any conda packages requested
-    condaInstructions.map(CondaEnvironmentManager.createOrGetCondaEnvironment)
+    condaInstructions.map(CondaEnvironmentManager.getOrCreateCondaEnvironment)
   }
 
   private[this] val envVars: Map[String, String] = {

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

@@ -348,7 +348,7 @@ private[r] object RRunner {
     val sparkConf = SparkEnv.get.conf
     val requestedRCommand = Provenance.fromConf("spark.r.command")
       .orElse(Provenance.fromConf("spark.sparkr.r.command"))
-    val condaEnv = condaSetupInstructions.map(CondaEnvironmentManager.createOrGetCondaEnvironment)
+    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"))