[Core] Allow users to specify the classpath and import path (#10560)

* move job resource path to job config

* job resource path support list

* job resource path support for python

* fix job_resource_path support

* fix worker command

* fix job config

* use jar file instead of parent path

* fix job resource path

* add test to test.sh

* lint

* Update java/runtime/src/main/resources/ray.default.conf

Co-authored-by: Kai Yang <kfstorm@outlook.com>

* fix testGetFunctionFromLocalResource

* lint

* fix rebase

* add jars in resource path to classloader

* add job_resource_path to worker

* add ray stop

* rename job_resource_path to resource_path

* fix resource_path

* refine resource_path comments

* rename job resource path to code search path

* Add instruction about starting a cross-language cluster

* fix ClassLoaderTest.java

* add code-search-path to RunManager

* refine comments for code-search-path

* rename resourcePath to codeSearchPath

* Update doc

* fix

* rename resourcePath to codeSearchPath

* update doc

* filter out empty path

* fix comments

* fix comments

* fix tests

* revert pom

* lint

* fix doc

* update doc

* Apply suggestions from code review

* lint

Co-authored-by: Kai Yang <kfstorm@outlook.com>
Co-authored-by: Hao Chen <chenh1024@gmail.com>

doc/source/cross-language.rst

@@ -3,6 +3,25 @@ Cross-language programming
 
 This page will show you how to use Ray's cross-language programming feature.
 
+Setup the cluster
+-----------------
+
+We need to set the ``--code-search-path`` option on ``ray start`` command. The ``--code-search-path`` option instructs workers to load Java or Python code from the specified code search path.
+
+.. code-block:: bash
+
+  ray start ... --code-search-path=/path/to/code
+
+You can also provide multiple directories for this option.
+
+.. code-block:: bash
+
+  ray start ... --code-search-path=/path/to/jars1:/path/to/jars2:/path/to/pys1:/path/to/pys2
+
+.. note:
+
+  If ``--code-search-path`` is specified, you can only run remote functions which can be found in ``--code-search-path``.
+
 Python calling Java
 -------------------
 
@@ -41,7 +60,7 @@ from the above Java class.
 
   import ray
 
-  ray.init(_load_code_from_local=True)
+  ray.init(address="auto")
 
   # Define a Java class.
   counter_class = ray.java_actor_class(
@@ -57,13 +76,13 @@ from the above Java class.
   # Define a Java function.
   add_function = ray.java_function(
         "io.ray.demo.Math", "add")
-  
+
   # Call the Java remote function.
   obj_ref3 = add_function.remote(1, 2)
   assert ray.get(obj_ref3) == 3
 
   ray.shutdown()
-  
+
 Java calling Python
 -------------------
 
@@ -92,7 +111,7 @@ Suppose we have a Python module as follows:
 
   * The function or class should be decorated by `@ray.remote`.
 
-Then, in Java, we can call the above Python remote function, or create an actor 
+Then, in Java, we can call the above Python remote function, or create an actor
 from the above Python class.
 
 .. code-block:: java
@@ -139,7 +158,7 @@ Cross-language data serialization
 
 The arguments and return values of ray call can be serialized & deserialized
 automatically if their types are the following:
-  
+
   - Primitive data types
       ===========   =======  =======
       MessagePack   Python   Java
@@ -168,7 +187,7 @@ automatically if their types are the following:
     float type to receive the input argument, the double precision Python data
     will be reduced to float precision in Java.
   * BigInteger can support max value of 2^64-1, please refer to:
-    https://github.com/msgpack/msgpack/blob/master/spec.md#int-format-family. 
+    https://github.com/msgpack/msgpack/blob/master/spec.md#int-format-family.
     If the value larger than 2^64-1, then transfer the BigInteger:
 
       - From Java to Python: *raise an exception*
@@ -272,7 +291,7 @@ Then, run the following code:
 
   import ray
 
-  ray.init(_load_code_from_local=True)
+  ray.init(address="auto")
 
   obj_ref = ray.java_function(
         "io.ray.demo.MyRayClass",

doc/source/starting-ray.rst

@@ -135,6 +135,8 @@ There are two steps needed to use Ray in a distributed setting:
 
           ray up cluster.yaml
 
+      To configure the Ray cluster to run Java code, you need to add the ``--code-search-path`` option. It's used to specify classpath for workers in the cluster. Your jar files must be distributed to all the nodes of the Ray cluster before running your code. You also need to make sure the paths of jar files are the same among nodes.
+
       You can monitor the Ray cluster status with ``ray monitor cluster.yaml`` and ssh into the head node with ``ray attach cluster.yaml``.
 
     2. Specify the address of the Ray cluster when initializing Ray in your code. This causes Ray to connect to the existing cluster instead of starting a new one on the local node.
@@ -152,14 +154,11 @@ There are two steps needed to use Ray in a distributed setting:
 
           You need to add the ``ray.redis.address`` parameter to your command line (like ``-Dray.redis.address=...``).
 
-          You need to add the ``ray.job.resource-path`` parameter as well. Your jar files must be distributed to all the nodes of the Ray cluster before running your code. You also need to make sure the paths of jar files are the same among nodes. Let's say your jar files are located in ``/path/to/jars/``, all files under this path will be loaded by worker processes.
-
           To connect your program to the Ray cluster, run it like this:
 
               .. code-block:: bash
 
                   java -classpath /path/to/jars/ \
-                    -Dray.job.resource-path=/path/to/jars/ \
                     -Dray.redis.address=<address> \
                     <classname> <args>
 
@@ -189,6 +188,11 @@ The command will print out the address of the Redis server that was started (and
 
     $ ray start --address=<address>
 
+If you want to run Java code, you need to specify the classpath via the ``--code-search-path`` option.
+
+.. code-block:: bash
+
+  $ ray start ... --code-search-path=/path/to/jars
 
 Local mode
 ----------

java/runtime/src/main/java/io/ray/runtime/AbstractRayRuntime.java

@@ -69,7 +69,7 @@ public abstract class AbstractRayRuntime implements RayRuntimeInternal {
   public AbstractRayRuntime(RayConfig rayConfig) {
     this.rayConfig = rayConfig;
     setIsContextSet(rayConfig.workerMode == Common.WorkerType.DRIVER);
-    functionManager = new FunctionManager(rayConfig.jobResourcePath);
+    functionManager = new FunctionManager(rayConfig.codeSearchPath);
     runtimeContext = new RuntimeContextImpl(this);
   }
 

java/runtime/src/main/java/io/ray/runtime/RayNativeRuntime.java

@@ -157,7 +157,8 @@ public final class RayNativeRuntime extends AbstractRayRuntime {
           JobConfig.newBuilder()
               .setNumJavaWorkersPerProcess(rayConfig.numWorkersPerProcess)
               .addAllJvmOptions(rayConfig.jvmOptionsForJavaWorker)
-              .putAllWorkerEnv(rayConfig.workerEnv);
+              .putAllWorkerEnv(rayConfig.workerEnv)
+              .addAllCodeSearchPath(rayConfig.codeSearchPath);
       serializedJobConfig = jobConfigBuilder.build().toByteArray();
     }
 

java/runtime/src/main/java/io/ray/runtime/config/RayConfig.java

@@ -16,6 +16,8 @@ import io.ray.runtime.util.ResourceUtil;
 import java.io.File;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -73,7 +75,7 @@ public class RayConfig {
   public int nodeManagerPort;
   public final Map<String, String> rayletConfigParameters;
 
-  public final String jobResourcePath;
+  public List<String> codeSearchPath;
   public final String pythonWorkerCommand;
 
   private static volatile RayConfig instance = null;
@@ -225,11 +227,12 @@ public class RayConfig {
       rayletConfigParameters.put(entry.getKey(), value == null ? "" : value.toString());
     }
 
-    // Job resource path.
-    if (config.hasPath("ray.job.resource-path")) {
-      jobResourcePath = config.getString("ray.job.resource-path");
+    // Job code search path.
+    if (config.hasPath("ray.job.code-search-path")) {
+      codeSearchPath = Arrays.asList(
+          config.getString("ray.job.code-search-path").split(":"));
     } else {
-      jobResourcePath = null;
+      codeSearchPath = Collections.emptyList();
     }
 
     boolean enableMultiTenancy = false;
@@ -311,7 +314,7 @@ public class RayConfig {
     dynamic.put("ray.object-store.socket-name", objectStoreSocketName);
     dynamic.put("ray.raylet.node-manager-port", nodeManagerPort);
     dynamic.put("ray.redis.address", redisAddress);
-    dynamic.put("ray.job.resource-path", jobResourcePath);
+    dynamic.put("ray.job.code-search-path", codeSearchPath);
     Config toRender = ConfigFactory.parseMap(dynamic).withFallback(config);
     return toRender.root().render(ConfigRenderOptions.concise());
   }

java/runtime/src/main/java/io/ray/runtime/functionmanager/FunctionManager.java

@@ -1,6 +1,5 @@
 package io.ray.runtime.functionmanager;
 
-import com.google.common.base.Strings;
 import io.ray.api.function.RayFunc;
 import io.ray.api.id.JobId;
 import io.ray.runtime.util.LambdaUtils;
@@ -12,6 +11,8 @@ import java.lang.reflect.Method;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -21,10 +22,11 @@ import java.util.Map;
 import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.filefilter.DirectoryFileFilter;
 import org.apache.commons.io.filefilter.RegexFileFilter;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
 import org.objectweb.asm.Type;
@@ -57,16 +59,16 @@ public class FunctionManager {
   /**
    * The resource path which we can load the job's jar resources.
    */
-  private final String jobResourcePath;
+  private final List<String> codeSearchPath;
 
   /**
-   * Construct a FunctionManager with the specified job resource path.
+   * Construct a FunctionManager with the specified code search path.
    *
-   * @param jobResourcePath The specified job resource that can store the job's
+   * @param codeSearchPath The specified job resource that can store the job's
    *                        resources.
    */
-  public FunctionManager(String jobResourcePath) {
-    this.jobResourcePath = jobResourcePath;
+  public FunctionManager(List<String> codeSearchPath) {
+    this.codeSearchPath = codeSearchPath;
   }
 
   /**
@@ -115,23 +117,35 @@ public class FunctionManager {
 
   private JobFunctionTable createJobFunctionTable(JobId jobId) {
     ClassLoader classLoader;
-    if (Strings.isNullOrEmpty(jobResourcePath)) {
+    if (codeSearchPath == null || codeSearchPath.isEmpty()) {
       classLoader = getClass().getClassLoader();
     } else {
-      File resourceDir = new File(jobResourcePath + "/" + jobId.toString() + "/");
-      Collection<File> files = FileUtils.listFiles(resourceDir,
-          new RegexFileFilter(".*\\.jar"), DirectoryFileFilter.DIRECTORY);
-      files.add(resourceDir);
-      final List<URL> urlList = files.stream().map(file -> {
-        try {
-          return file.toURI().toURL();
-        } catch (MalformedURLException e) {
-          throw new RuntimeException(e);
-        }
-      }).collect(Collectors.toList());
-      classLoader = new URLClassLoader(urlList.toArray(new URL[urlList.size()]));
-      LOGGER.debug("Resource loaded for job {} from path {}.", jobId,
-          resourceDir.getAbsolutePath());
+      URL[] urls = codeSearchPath.stream()
+          .filter(p -> StringUtils.isNotBlank(p) && Files.exists(Paths.get(p)))
+          .flatMap(p -> {
+            try {
+              if (!Files.isDirectory(Paths.get(p))) {
+                if (!p.endsWith(".jar")) {
+                  return Stream.of(Paths.get(p).getParent().toAbsolutePath().toUri().toURL());
+                } else {
+                  return Stream.of(Paths.get(p).toAbsolutePath().toUri().toURL());
+                }
+              } else {
+                List<URL> subUrls = new ArrayList<>();
+                subUrls.add(Paths.get(p).toAbsolutePath().toUri().toURL());
+                Collection<File> jars = FileUtils.listFiles(new File(p),
+                    new RegexFileFilter(".*\\.jar"), DirectoryFileFilter.DIRECTORY);
+                for (File jar : jars) {
+                  subUrls.add(jar.toPath().toUri().toURL());
+                }
+                return subUrls.stream();
+              }
+            } catch (MalformedURLException e) {
+              throw new RuntimeException(String.format("Illegal %s resource path", p));
+            }
+          }).toArray(URL[]::new);
+      classLoader = new URLClassLoader(urls);
+      LOGGER.debug("Resource loaded for job {} from path {}.", jobId, urls);
     }
 
     return new JobFunctionTable(classLoader);

java/runtime/src/main/java/io/ray/runtime/runner/RunManager.java

@@ -357,7 +357,10 @@ public class RunManager {
     File workerConfigFile = new File(rayConfig.sessionDir + "/java_worker.conf");
     FileUtils.write(workerConfigFile, rayConfig.render(), Charset.defaultCharset());
     cmd.add("-Dray.config-file=" + workerConfigFile.getAbsolutePath());
-
+    if (!rayConfig.codeSearchPath.isEmpty()) {
+      cmd.add("-Dray.job.code-search-path=" +
+          String.join(":", rayConfig.codeSearchPath));
+    }
     cmd.add("RAY_WORKER_RAYLET_CONFIG_PLACEHOLDER");
 
     cmd.addAll(rayConfig.jvmParameters);

java/runtime/src/main/resources/ray.default.conf

@@ -22,10 +22,10 @@ ray {
     // If worker.mode is DRIVER, specify the job id.
     // If not provided, a random id will be used.
     id: ""
-    // If this config is set, worker will use different paths to load resources when
-    // executing tasks from different jobs. E.g. if it's set to '/tm/job_resources',
-    // the path for job 123 will be '/tmp/job_resources/123'.
-    resource-path: ""
+    // A list of directories or jar files separated by colon that specify the
+    // search path for user code. This will be used as `CLASSPATH` in Java,
+    // and `PYTHONPATH` in Python.
+    code-search-path: ""
     /// The number of java worker per worker process.
     num-java-workers-per-process: 1
     /// The jvm options for java workers of the job.

java/runtime/src/test/java/io/ray/runtime/config/RayConfigTest.java

@@ -1,6 +1,7 @@
 package io.ray.runtime.config;
 
 import io.ray.runtime.generated.Common.WorkerType;
+import java.util.Collections;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -11,13 +12,14 @@ public class RayConfigTest {
   @Test
   public void testCreateRayConfig() {
     try {
-      System.setProperty("ray.job.resource-path", "path/to/ray/job/resource/path");
+      System.setProperty("ray.job.code-search-path", "path/to/ray/job/resource/path");
       RayConfig rayConfig = RayConfig.create();
       Assert.assertEquals(WorkerType.DRIVER, rayConfig.workerMode);
-      Assert.assertEquals("path/to/ray/job/resource/path", rayConfig.jobResourcePath);
+      Assert.assertEquals(Collections.singletonList("path/to/ray/job/resource/path"),
+          rayConfig.codeSearchPath);
     } finally {
       // Unset system properties.
-      System.clearProperty("ray.job.resource-path");
+      System.clearProperty("ray.job.code-search-path");
     }
   }
 

java/runtime/src/test/java/io/ray/runtime/functionmanager/FunctionManagerTest.java

@@ -7,6 +7,7 @@ import io.ray.runtime.functionmanager.FunctionManager.JobFunctionTable;
 import java.io.File;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.Collections;
 import java.util.Map;
 import javax.tools.JavaCompiler;
 import javax.tools.ToolProvider;
@@ -151,9 +152,8 @@ public class FunctionManagerTest {
   @Test
   public void testGetFunctionFromLocalResource() throws Exception {
     JobId jobId = JobId.fromInt(1);
-    final String resourcePath = FileUtils.getTempDirectoryPath() + "/ray_test_resources";
-    final String jobResourcePath = resourcePath + "/" + jobId.toString();
-    File jobResourceDir = new File(jobResourcePath);
+    final String codeSearchPath = FileUtils.getTempDirectoryPath() + "/ray_test_resources/";
+    File jobResourceDir = new File(codeSearchPath);
     FileUtils.deleteQuietly(jobResourceDir);
     jobResourceDir.mkdirs();
     jobResourceDir.deleteOnExit();
@@ -165,13 +165,13 @@ public class FunctionManagerTest {
     demoJavaFile += "  }\n";
     demoJavaFile += "}";
 
-    // Write the demo java file to the job resource path.
-    String javaFilePath = jobResourcePath + "/DemoApp.java";
+    // Write the demo java file to the job code search path.
+    String javaFilePath = codeSearchPath + "/DemoApp.java";
     Files.write(Paths.get(javaFilePath), demoJavaFile.getBytes());
 
     // Compile the java file.
     JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
-    int result = compiler.run(null, null, null, "-d", jobResourcePath, javaFilePath);
+    int result = compiler.run(null, null, null, "-d", codeSearchPath, javaFilePath);
     if (result != 0) {
       throw new RuntimeException("Couldn't compile Demo.java.");
     }
@@ -179,7 +179,8 @@ public class FunctionManagerTest {
     // Test loading the function.
     JavaFunctionDescriptor descriptor = new JavaFunctionDescriptor(
         "DemoApp", "hello", "()Ljava/lang/String;");
-    final FunctionManager functionManager = new FunctionManager(resourcePath);
+    final FunctionManager functionManager = new FunctionManager(
+        Collections.singletonList(codeSearchPath));
     RayFunction func = functionManager.getFunction(jobId, descriptor);
     Assert.assertEquals(func.getFunctionDescriptor(), descriptor);
   }

java/test.sh

@@ -46,6 +46,16 @@ echo "Running tests under single-process mode."
 # bazel test //java:all_tests --jvmopt="-Dray.run-mode=SINGLE_PROCESS" --config=ci || single_exit_code=$?
 run_testng java -Dray.run-mode="SINGLE_PROCESS" -cp "$ROOT_DIR"/../bazel-bin/java/all_tests_deploy.jar "${TEST_ARGS[@]}" org.testng.TestNG -d /tmp/ray_java_test_output "$ROOT_DIR"/testng.xml
 
+echo "Running connecting existing cluster tests"
+case "${OSTYPE}" in
+  linux*) ip=$(hostname -I | awk '{print $1}');;
+  darwin*) ip=$(ipconfig getifaddr en0);;
+  *) echo "Can't get ip address for ${OSTYPE}"; exit 1;;
+esac
+RAY_BACKEND_LOG_LEVEL=debug ray start --head --redis-port=6379 --redis-password=123456 --include-java --code-search-path="$PWD/bazel-bin/java/all_tests_deploy.jar"
+RAY_BACKEND_LOG_LEVEL=debug java -cp bazel-bin/java/all_tests_deploy.jar -Dray.redis.address="$ip:6379"\
+ -Dray.redis.password='123456' -Dray.job.code-search-path="$PWD/bazel-bin/java/all_tests_deploy.jar" io.ray.test.MultiDriverTest
+ray stop
 popd
 
 pushd "$ROOT_DIR"

java/test/src/main/java/io/ray/test/BaseMultiLanguageTest.java

@@ -8,6 +8,7 @@ import io.ray.runtime.config.RayConfig;
 import io.ray.runtime.util.NetworkUtil;
 import java.io.File;
 import java.lang.ProcessBuilder.Redirect;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -66,12 +67,11 @@ public abstract class BaseMultiLanguageTest {
     // jars in the `ray` wheel doesn't contains test classes, so we add test classes explicitly.
     // Since mvn test classes contains `test` in path and bazel test classes is located at a jar
     // with `test` included in the name, we can check classpath `test` to filter out test classes.
-    String classpath = Stream.of(System.getProperty("java.class.path").split(":"))
+    List<String> classpath = Stream.of(System.getProperty("java.class.path").split(":"))
         .filter(s -> !s.contains(" ") && s.contains("test"))
-        .collect(Collectors.joining(":"));
-    String workerOptions = new Gson().toJson(ImmutableList.of("-classpath", classpath));
+        .collect(Collectors.toList());
     // Start ray cluster.
-    List<String> startCommand = ImmutableList.of(
+    List<String> startCommand = Arrays.asList(
         "ray",
         "start",
         "--head",
@@ -83,9 +83,10 @@ public abstract class BaseMultiLanguageTest {
         String.format("--node-manager-port=%s", nodeManagerPort),
         "--load-code-from-local",
         "--include-java",
-        "--java-worker-options=" + workerOptions,
-        "--system-config=" + new Gson().toJson(RayConfig.create().rayletConfigParameters)
+        "--system-config=" + new Gson().toJson(RayConfig.create().rayletConfigParameters),
+        "--code-search-path=" + String.join(":", classpath)
     );
+
     if (!executeCommand(startCommand, 10, getRayStartEnv())) {
       throw new RuntimeException("Couldn't start ray cluster.");
     }

java/test/src/main/java/io/ray/test/ClassLoaderTest.java

@@ -3,7 +3,6 @@ package io.ray.test;
 import io.ray.api.ActorHandle;
 import io.ray.api.BaseActorHandle;
 import io.ray.api.ObjectRef;
-import io.ray.api.Ray;
 import io.ray.api.options.ActorCreationOptions;
 import io.ray.runtime.AbstractRayRuntime;
 import io.ray.runtime.functionmanager.FunctionDescriptor;
@@ -23,31 +22,31 @@ import org.testng.annotations.Test;
 
 public class ClassLoaderTest extends BaseTest {
 
-  private final String resourcePath = FileUtils.getTempDirectoryPath()
+  private final String codeSearchPath = FileUtils.getTempDirectoryPath()
       + "/ray_test/ClassLoaderTest";
 
   @BeforeClass
   public void setUp() {
     // The potential issue of multiple `ClassLoader` instances for the same job on multi-threading
-    // scenario only occurs if the classes are loaded from the job resource path.
-    System.setProperty("ray.job.resource-path", resourcePath);
+    // scenario only occurs if the classes are loaded from the job code search path.
+    System.setProperty("ray.job.code-search-path", codeSearchPath);
   }
 
   @AfterClass
   public void tearDown() {
-    System.clearProperty("ray.job.resource-path");
+    System.clearProperty("ray.job.code-search-path");
   }
 
   @Test(groups = {"cluster"})
   public void testClassLoaderInMultiThreading() throws Exception {
-    final String jobResourcePath = resourcePath + "/" + Ray.getRuntimeContext().getCurrentJobId();
-    File jobResourceDir = new File(jobResourcePath);
+    File jobResourceDir = new File(codeSearchPath);
     FileUtils.deleteQuietly(jobResourceDir);
     jobResourceDir.mkdirs();
     jobResourceDir.deleteOnExit();
 
-    // In this test case the class is expected to be loaded from the job resource path, so we need
-    // to put the compiled class file into the job resource path and load it later.
+    // In this test case the class is expected to be loaded from the job code search path,
+    // so we need to put the compiled class file into the job code search path and load it
+    // later.
     String testJavaFile = ""
         + "import java.lang.management.ManagementFactory;\n"
         + "import java.lang.management.RuntimeMXBean;\n"
@@ -83,14 +82,14 @@ public class ClassLoaderTest extends BaseTest {
         + "  }\n"
         + "}";
 
-    // Write the demo java file to the job resource path.
-    String javaFilePath = jobResourcePath + "/ClassLoaderTester.java";
+    // Write the demo java file to the job code search path.
+    String javaFilePath = codeSearchPath + "/ClassLoaderTester.java";
     Files.write(Paths.get(javaFilePath), testJavaFile.getBytes());
 
     // Compile the java file.
     JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
     int result = compiler.run(null, null, null, "-d",
-        jobResourcePath, javaFilePath);
+        codeSearchPath, javaFilePath);
     if (result != 0) {
       throw new RuntimeException("Couldn't compile ClassLoaderTester.java.");
     }

python/ray/job_config.py

@@ -10,6 +10,9 @@ class JobConfig:
         num_java_workers_per_process (int): The number of java workers per
             worker process.
         jvm_options (str[]): The jvm options for java workers of the job.
+        code_search_path (list): A list of directories or jar files that
+            specify the search path for user code. This will be used as
+            `CLASSPATH` in Java and `PYTHONPATH` in Python.
     """
 
     def __init__(
@@ -17,6 +20,7 @@ class JobConfig:
             worker_env=None,
             num_java_workers_per_process=1,
             jvm_options=None,
+            code_search_path=None,
     ):
         if worker_env is None:
             self.worker_env = dict()
@@ -27,6 +31,8 @@ class JobConfig:
             self.jvm_options = []
         else:
             self.jvm_options = jvm_options
+        if code_search_path is None:
+            self.code_search_path = []
 
     def serialize(self):
         job_config = ray.gcs_utils.JobConfig()
@@ -35,4 +41,5 @@ class JobConfig:
         job_config.num_java_workers_per_process = (
             self.num_java_workers_per_process)
         job_config.jvm_options.extend(self.jvm_options)
+        job_config.code_search_path.extend(self.code_search_path)
         return job_config.SerializeToString()

python/ray/node.py

@@ -730,7 +730,8 @@ class Node:
             head_node=self.head,
             start_initial_python_workers_for_first_job=self._ray_params.
             start_initial_python_workers_for_first_job,
-            object_spilling_config=self._ray_params.object_spilling_config)
+            object_spilling_config=self._ray_params.object_spilling_config,
+            code_search_path=self._ray_params.code_search_path)
         assert ray_constants.PROCESS_TYPE_RAYLET not in self.all_processes
         self.all_processes[ray_constants.PROCESS_TYPE_RAYLET] = [process_info]
 

python/ray/parameter.py

@@ -147,7 +147,8 @@ class RayParams:
                  metrics_agent_port=None,
                  metrics_export_port=None,
                  lru_evict=False,
-                 object_spilling_config=None):
+                 object_spilling_config=None,
+                 code_search_path=None):
         self.object_ref_seed = object_ref_seed
         self.redis_address = redis_address
         self.num_cpus = num_cpus
@@ -194,6 +195,9 @@ class RayParams:
         self._enable_object_reconstruction = enable_object_reconstruction
         self.object_spilling_config = object_spilling_config
         self._check_usage()
+        self.code_search_path = code_search_path
+        if code_search_path is None:
+            self.code_search_path = []
 
         # Set the internal config options for LRU eviction.
         if lru_evict:

python/ray/scripts/scripts.py

@@ -354,6 +354,13 @@ def dashboard(cluster_config_file, cluster_name, port, remote_port):
     default=None,
     type=str,
     help="Overwrite the options to start Java workers.")
+@click.option(
+    "--code-search-path",
+    default=None,
+    type=str,
+    help="A list of directories or jar files separated by colon that specify "
+    "the search path for user code. This will be used as `CLASSPATH` in "
+    "Java and `PYTHONPATH` in Python.")
 @click.option(
     "--system-config",
     default=None,
@@ -391,9 +398,9 @@ def start(node_ip_address, redis_address, address, redis_port, port,
           dashboard_port, block, plasma_directory, huge_pages,
           autoscaling_config, no_redirect_worker_output, no_redirect_output,
           plasma_store_socket_name, raylet_socket_name, temp_dir, include_java,
-          java_worker_options, load_code_from_local, system_config, lru_evict,
-          enable_object_reconstruction, metrics_export_port, log_style,
-          log_color, verbose):
+          java_worker_options, code_search_path, load_code_from_local,
+          system_config, lru_evict, enable_object_reconstruction,
+          metrics_export_port, log_style, log_color, verbose):
     """Start Ray processes manually on the local machine."""
     cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
@@ -504,6 +511,7 @@ def start(node_ip_address, redis_address, address, redis_port, port,
         dashboard_port=dashboard_port,
         java_worker_options=java_worker_options,
         load_code_from_local=load_code_from_local,
+        code_search_path=code_search_path,
         _system_config=system_config,
         lru_evict=lru_evict,
         enable_object_reconstruction=enable_object_reconstruction,

python/ray/services.py

@@ -11,16 +11,15 @@ import socket
 import subprocess
 import sys
 import time
-import redis
 
 import colorama
+import colorful as cf
+import psutil
 # Ray modules
 import ray
 import ray.ray_constants as ray_constants
-import psutil
-
+import redis
 from ray.autoscaler.cli_logger import cli_logger
-import colorful as cf
 
 resource = None
 if sys.platform != "win32":
@@ -71,11 +70,6 @@ DEFAULT_WORKER_EXECUTABLE = os.path.join(
     os.path.abspath(os.path.dirname(__file__)),
     "core/src/ray/cpp/default_worker" + EXE_SUFFIX)
 
-DEFAULT_JAVA_WORKER_CLASSPATH = [
-    os.path.join(
-        os.path.abspath(os.path.dirname(__file__)), "../../../build/java/*"),
-]
-
 # Logger for this module. It should be configured at the entry point
 # into the program using Ray. Ray provides a default configuration at
 # entry/init points.
@@ -1282,7 +1276,8 @@ def start_raylet(redis_address,
                  socket_to_use=None,
                  head_node=False,
                  start_initial_python_workers_for_first_job=False,
-                 object_spilling_config=None):
+                 object_spilling_config=None,
+                 code_search_path=None):
     """Start a raylet, which is a combined local scheduler and object manager.
 
     Args:
@@ -1320,6 +1315,9 @@ def start_raylet(redis_address,
         include_java (bool): If True, the raylet backend can also support
             Java worker.
         java_worker_options (list): The command options for Java worker.
+        code_search_path (list): Code search path for worker. code_search_path
+            is added to worker command in non-multi-tenancy mode and job_config
+            in multi-tenancy mode.
     Returns:
         ProcessInfo for the process that was started.
     """
@@ -1348,16 +1346,15 @@ def start_raylet(redis_address,
     gcs_ip_address, gcs_port = redis_address.split(":")
 
     if include_java is True:
-        default_cp = os.pathsep.join(DEFAULT_JAVA_WORKER_CLASSPATH)
         java_worker_command = build_java_worker_command(
-            json.loads(java_worker_options)
-            if java_worker_options else ["-classpath", default_cp],
+            json.loads(java_worker_options) if java_worker_options else [],
             redis_address,
             node_manager_port,
             plasma_store_name,
             raylet_name,
             redis_password,
             session_dir,
+            code_search_path,
         )
     else:
         java_worker_command = []
@@ -1384,6 +1381,8 @@ def start_raylet(redis_address,
         f"--config-list={config_str}", f"--temp-dir={temp_dir}",
         f"--metrics-agent-port={metrics_agent_port}"
     ]
+    if code_search_path:
+        start_worker_command.append(f"--code-search-path={code_search_path}")
     if redis_password:
         start_worker_command += [f"--redis-password={redis_password}"]
 
@@ -1398,6 +1397,9 @@ def start_raylet(redis_address,
     if max_worker_port is None:
         max_worker_port = 0
 
+    if code_search_path is not None and len(code_search_path) > 0:
+        load_code_from_local = True
+
     if load_code_from_local:
         start_worker_command += ["--load-code-from-local"]
 
@@ -1493,15 +1495,10 @@ def get_ray_jars_dir():
     return os.path.abspath(os.path.join(current_dir, "jars"))
 
 
-def build_java_worker_command(
-        java_worker_options,
-        redis_address,
-        node_manager_port,
-        plasma_store_name,
-        raylet_name,
-        redis_password,
-        session_dir,
-):
+def build_java_worker_command(java_worker_options, redis_address,
+                              node_manager_port, plasma_store_name,
+                              raylet_name, redis_password, session_dir,
+                              code_search_path):
     """This method assembles the command used to start a Java worker.
 
     Args:
@@ -1512,6 +1509,7 @@ def build_java_worker_command(
         raylet_name (str): The name of the raylet socket to create.
         redis_password (str): The password of connect to redis.
         session_dir (str): The path of this session.
+        code_search_path (list): Teh job code search path.
     Returns:
         The command string for starting Java worker.
     """
@@ -1532,7 +1530,7 @@ def build_java_worker_command(
     pairs.append(("ray.home", RAY_HOME))
     pairs.append(("ray.logging.dir", os.path.join(session_dir, "logs")))
     pairs.append(("ray.session-dir", session_dir))
-
+    pairs.append(("ray.job.code-search-path", code_search_path))
     command = ["java"] + ["-D{}={}".format(*pair) for pair in pairs]
 
     command += ["RAY_WORKER_RAYLET_CONFIG_PLACEHOLDER"]

python/ray/worker.py

@@ -493,6 +493,7 @@ def init(
         _redis_password=ray_constants.REDIS_DEFAULT_PASSWORD,
         _include_java=False,
         _java_worker_options=None,
+        _code_search_path=None,
         _temp_dir=None,
         _load_code_from_local=False,
         _lru_evict=False,
@@ -587,6 +588,7 @@ def init(
         _load_code_from_local: Whether code should be loaded from a local
             module or from the GCS.
         _java_worker_options: Overwrite the options to start Java workers.
+        _code_search_path (list): Java classpath or python import path.
         _lru_evict (bool): If True, when an object store is full, it will evict
             objects in LRU order to make more space and when under memory
             pressure, ray.ObjectLostError may be thrown. If False, then
@@ -682,6 +684,7 @@ def init(
             temp_dir=_temp_dir,
             load_code_from_local=_load_code_from_local,
             java_worker_options=_java_worker_options,
+            code_search_path=_code_search_path,
             start_initial_python_workers_for_first_job=True,
             _system_config=_system_config,
             lru_evict=_lru_evict,

python/ray/workers/default_worker.py

@@ -1,6 +1,8 @@
 import argparse
 import json
 import time
+import sys
+import os
 
 import ray
 import ray.actor
@@ -98,7 +100,13 @@ parser.add_argument(
     type=str,
     default="",
     help="The configuration of object spilling. Only used by I/O workers.")
-
+parser.add_argument(
+    "--code-search-path",
+    default=None,
+    type=str,
+    help="A list of directories or jar files separated by colon that specify "
+    "the search path for user code. This will be used as `CLASSPATH` in "
+    "Java and `PYTHONPATH` in Python.")
 if __name__ == "__main__":
     args = parser.parse_args()
 
@@ -135,6 +143,13 @@ if __name__ == "__main__":
     if raylet_ip_address is None:
         raylet_ip_address = args.node_ip_address
 
+    code_search_path = args.code_search_path
+    if code_search_path is not None:
+        for p in code_search_path.split(":"):
+            if os.path.isfile(p):
+                p = os.path.dirname(p)
+            sys.path.append(p)
+
     ray_params = RayParams(
         node_ip_address=args.node_ip_address,
         raylet_ip_address=raylet_ip_address,

src/ray/protobuf/gcs.proto

@@ -321,6 +321,10 @@ message JobConfig {
   uint32 num_java_workers_per_process = 2;
   // The jvm options for java workers of the job.
   repeated string jvm_options = 3;
+  // A list of directories or jar files that specify the search path for user
+  // code. This will be used as `CLASSPATH` in Java, and `PYTHONPATH` in
+  // Python.
+  repeated string code_search_path = 4;
 }
 
 message JobTableData {

src/ray/raylet/worker_pool.cc

@@ -224,6 +224,33 @@ Process WorkerPool::StartWorkerProcess(const Language &language,
     dynamic_options.insert(dynamic_options.begin(), job_config->jvm_options().begin(),
                            job_config->jvm_options().end());
   }
+  // For non-multi-tenancy mode, job code search path is embedded in worker_command.
+  if (RayConfig::instance().enable_multi_tenancy() && job_config) {
+    std::string code_search_path_str;
+    for (int i = 0; i < job_config->code_search_path_size(); i++) {
+      auto path = job_config->code_search_path(i);
+      if (i != 0) {
+        code_search_path_str += ":";
+      }
+      code_search_path_str += path;
+    }
+    if (!code_search_path_str.empty()) {
+      switch (language) {
+      case Language::PYTHON: {
+        code_search_path_str = "--code-search-path=" + code_search_path_str;
+        break;
+      }
+      case Language::JAVA: {
+        code_search_path_str = "-Dray.job.code-search-path" + code_search_path_str;
+        break;
+      }
+      default:
+        RAY_LOG(FATAL) << "code_search_path is not supported for worker language "
+                       << language;
+      }
+      dynamic_options.push_back(code_search_path_str);
+    }
+  }
 
   // Extract pointers from the worker command to pass into execvp.
   std::vector<std::string> worker_command_args;