diff --git a/java/api/src/main/java/io/ray/api/Ray.java b/java/api/src/main/java/io/ray/api/Ray.java index 02ffc59c8..da9047a66 100644 --- a/java/api/src/main/java/io/ray/api/Ray.java +++ b/java/api/src/main/java/io/ray/api/Ray.java @@ -51,7 +51,7 @@ public final class Ray extends RayCall { /** * Check if {@link #init} has been called yet. * - *

Returns True if {@link #init} has already been called and false otherwise. + * @return True if {@link #init} has already been called and false otherwise. */ public static boolean isInitialized() { return runtime != null; @@ -60,8 +60,8 @@ public final class Ray extends RayCall { /** * Store an object in the object store. * - * @param obj The Java object to be stored. Returns A ObjectRef instance that represents the - * in-store object. + * @param obj The Java object to be stored. + * @return A ObjectRef instance that represents the in-store object. */ public static ObjectRef put(T obj) { return internal().put(obj); @@ -70,7 +70,8 @@ public final class Ray extends RayCall { /** * Get an object by `ObjectRef` from the object store. * - * @param objectRef The reference of the object to get. Returns The Java object. + * @param objectRef The reference of the object to get. + * @return The Java object. */ public static T get(ObjectRef objectRef) { return internal().get(objectRef); @@ -79,7 +80,8 @@ public final class Ray extends RayCall { /** * Get a list of objects by `ObjectRef`s from the object store. * - * @param objectList A list of object references. Returns A list of Java objects. + * @param objectList A list of object references. + * @return A list of Java objects. */ public static List get(List> objectList) { return internal().get(objectList); @@ -91,8 +93,8 @@ public final class Ray extends RayCall { * * @param waitList A list of object references to wait for. * @param numReturns The number of objects that should be returned. - * @param timeoutMs The maximum time in milliseconds to wait before returning. Returns Two lists, - * one containing locally available objects, one containing the rest. + * @param timeoutMs The maximum time in milliseconds to wait before returning. + * @return Two lists, one containing locally available objects, one containing the rest. */ public static WaitResult wait(List> waitList, int numReturns, int timeoutMs) { return internal().wait(waitList, numReturns, timeoutMs); @@ -103,8 +105,8 @@ public final class Ray extends RayCall { * objects are locally available. * * @param waitList A list of object references to wait for. - * @param numReturns The number of objects that should be returned. Returns Two lists, one - * containing locally available objects, one containing the rest. + * @param numReturns The number of objects that should be returned. + * @return Two lists, one containing locally available objects, one containing the rest. */ public static WaitResult wait(List> waitList, int numReturns) { return internal().wait(waitList, numReturns, Integer.MAX_VALUE); @@ -114,8 +116,8 @@ public final class Ray extends RayCall { * A convenient helper method for Ray.wait. It will wait infinitely until all objects are locally * available. * - * @param waitList A list of object references to wait for. Returns Two lists, one containing - * locally available objects, one containing the rest. + * @param waitList A list of object references to wait for. + * @return Two lists, one containing locally available objects, one containing the rest. */ public static WaitResult wait(List> waitList) { return internal().wait(waitList, waitList.size(), Integer.MAX_VALUE); @@ -127,8 +129,9 @@ public final class Ray extends RayCall { *

Gets a handle to a named actor with the given name. The actor must have been created with * name specified. * - * @param name The name of the named actor. Returns an ActorHandle to the actor if the actor of - * specified name exists or an Optional.empty() + * @param name The name of the named actor. + * @return an ActorHandle to the actor if the actor of specified name exists or an + * Optional.empty() */ public static Optional getActor(String name) { return internal().getActor(name, false); @@ -140,8 +143,9 @@ public final class Ray extends RayCall { *

Gets a handle to a global named actor with the given name. The actor must have been created * with global name specified. * - * @param name The global name of the named actor. Returns an ActorHandle to the actor if the - * actor of specified name exists or an Optional.empty() + * @param name The global name of the named actor. + * @return an ActorHandle to the actor if the actor of specified name exists or an + * Optional.empty() */ public static Optional getGlobalActor(String name) { return internal().getActor(name, true); @@ -151,7 +155,7 @@ public final class Ray extends RayCall { * If users want to use Ray API in their own threads, call this method to get the async context * and then call {@link #setAsyncContext} at the beginning of the new thread. * - *

Returns The async context. + * @return The async context. */ public static Object getAsyncContext() { return internal().getAsyncContext(); @@ -175,7 +179,8 @@ public final class Ray extends RayCall { * If users want to use Ray API in their own threads, they should wrap their {@link Runnable} * objects with this method. * - * @param runnable The runnable to wrap. Returns The wrapped runnable. + * @param runnable The runnable to wrap. + * @return The wrapped runnable. */ public static Runnable wrapRunnable(Runnable runnable) { return internal().wrapRunnable(runnable); @@ -185,7 +190,8 @@ public final class Ray extends RayCall { * If users want to use Ray API in their own threads, they should wrap their {@link Callable} * objects with this method. * - * @param callable The callable to wrap. Returns The wrapped callable. + * @param callable The callable to wrap. + * @return The wrapped callable. */ public static Callable wrapCallable(Callable callable) { return internal().wrapCallable(callable); @@ -238,7 +244,8 @@ public final class Ray extends RayCall { * * @param name Name of the placement group. * @param bundles Pre-allocated resource list. - * @param strategy Actor placement strategy. Returns A handle to the created placement group. + * @param strategy Actor placement strategy. + * @return A handle to the created placement group. */ public static PlacementGroup createPlacementGroup( String name, List> bundles, PlacementStrategy strategy) { @@ -265,7 +272,8 @@ public final class Ray extends RayCall { /** * Get a placement group by placement group Id. * - * @param id placement group id. Returns The placement group. + * @param id placement group id. + * @return The placement group. */ public static PlacementGroup getPlacementGroup(PlacementGroupId id) { return internal().getPlacementGroup(id); @@ -274,7 +282,7 @@ public final class Ray extends RayCall { /** * Get all placement groups in this cluster. * - *

Returns All placement groups. + * @return All placement groups. */ public static List getAllPlacementGroups() { return internal().getAllPlacementGroups(); diff --git a/java/api/src/main/java/io/ray/api/call/ActorCreator.java b/java/api/src/main/java/io/ray/api/call/ActorCreator.java index c6bb9cce8..b64a4fbcd 100644 --- a/java/api/src/main/java/io/ray/api/call/ActorCreator.java +++ b/java/api/src/main/java/io/ray/api/call/ActorCreator.java @@ -23,7 +23,8 @@ public class ActorCreator extends BaseActorCreator> { * *

Note, if this is set, this actor won't share Java worker with other actors or tasks. * - * @param jvmOptions JVM options for the Java worker that this actor is running in. Returns self + * @param jvmOptions JVM options for the Java worker that this actor is running in. + * @return self * @see io.ray.api.options.ActorCreationOptions.Builder#setJvmOptions(java.lang.String) */ public ActorCreator setJvmOptions(String jvmOptions) { @@ -34,7 +35,7 @@ public class ActorCreator extends BaseActorCreator> { /** * Create a java actor remotely and return a handle to the created actor. * - *

Returns a handle to the created java actor. + * @return a handle to the created java actor. */ public ActorHandle remote() { return Ray.internal().createActor(func, args, buildOptions()); diff --git a/java/api/src/main/java/io/ray/api/call/ActorTaskCaller.java b/java/api/src/main/java/io/ray/api/call/ActorTaskCaller.java index 4b9d25a21..4579acbb8 100644 --- a/java/api/src/main/java/io/ray/api/call/ActorTaskCaller.java +++ b/java/api/src/main/java/io/ray/api/call/ActorTaskCaller.java @@ -25,7 +25,7 @@ public class ActorTaskCaller { * Execute an java actor method remotely and return an object reference to the result object in * the object store. * - *

Returns an object reference to an object in the object store. + * @return an object reference to an object in the object store. */ @SuppressWarnings("unchecked") public ObjectRef remote() { diff --git a/java/api/src/main/java/io/ray/api/call/BaseActorCreator.java b/java/api/src/main/java/io/ray/api/call/BaseActorCreator.java index 5f488124b..7e761b4c2 100644 --- a/java/api/src/main/java/io/ray/api/call/BaseActorCreator.java +++ b/java/api/src/main/java/io/ray/api/call/BaseActorCreator.java @@ -18,7 +18,8 @@ public class BaseActorCreator { * name via {@link Ray#getActor(java.lang.String)}. If you want create a named actor that is * accessible from all jobs, use {@link BaseActorCreator#setGlobalName(java.lang.String)} instead. * - * @param name The name of the named actor. Returns self + * @param name The name of the named actor. + * @return self * @see io.ray.api.options.ActorCreationOptions.Builder#setName(String) */ public T setName(String name) { @@ -31,7 +32,8 @@ public class BaseActorCreator { * Ray#getGlobalActor(java.lang.String)}. If you want to create a named actor that is only * accessible from this job, use {@link BaseActorCreator#setName(java.lang.String)} instead. * - * @param name The name of the named actor. Returns self + * @param name The name of the named actor. + * @return self * @see io.ray.api.options.ActorCreationOptions.Builder#setGlobalName(String) */ public T setGlobalName(String name) { @@ -45,7 +47,8 @@ public class BaseActorCreator { * used. * * @param resourceName resource name - * @param resourceQuantity resource quantity Returns self + * @param resourceQuantity resource quantity + * @return self * @see ActorCreationOptions.Builder#setResource(java.lang.String, java.lang.Double) */ public T setResource(String resourceName, Double resourceQuantity) { @@ -58,7 +61,8 @@ public class BaseActorCreator { * called multiple times. If the same resource is set multiple times, the latest quantity will be * used. * - * @param resources requirements for multiple resources. Returns self + * @param resources requirements for multiple resources. + * @return self * @see BaseActorCreator#setResources(java.util.Map) */ public T setResources(Map resources) { @@ -71,7 +75,8 @@ public class BaseActorCreator { * unexpectedly. The minimum valid value is 0 (default), which indicates that the actor doesn't * need to be restarted. A value of -1 indicates that an actor should be restarted indefinitely. * - * @param maxRestarts max number of actor restarts Returns self + * @param maxRestarts max number of actor restarts + * @return self * @see ActorCreationOptions.Builder#setMaxRestarts(int) */ public T setMaxRestarts(int maxRestarts) { @@ -85,7 +90,8 @@ public class BaseActorCreator { *

The max concurrency defaults to 1 for threaded execution. Note that the execution order is * not guaranteed when {@code max_concurrency > 1}. * - * @param maxConcurrency The max number of concurrent calls to allow for this actor. Returns self + * @param maxConcurrency The max number of concurrent calls to allow for this actor. + * @return self * @see ActorCreationOptions.Builder#setMaxConcurrency(int) */ public T setMaxConcurrency(int maxConcurrency) { @@ -97,7 +103,8 @@ public class BaseActorCreator { * Set the placement group to place this actor in. * * @param group The placement group of the actor. - * @param bundleIndex The index of the bundle to place this actor in. Returns self + * @param bundleIndex The index of the bundle to place this actor in. + * @return self * @see ActorCreationOptions.Builder#setPlacementGroup(PlacementGroup, int) */ public T setPlacementGroup(PlacementGroup group, int bundleIndex) { diff --git a/java/api/src/main/java/io/ray/api/call/BaseTaskCaller.java b/java/api/src/main/java/io/ray/api/call/BaseTaskCaller.java index 8b683c7bd..88c58e053 100644 --- a/java/api/src/main/java/io/ray/api/call/BaseTaskCaller.java +++ b/java/api/src/main/java/io/ray/api/call/BaseTaskCaller.java @@ -14,7 +14,8 @@ public class BaseTaskCaller> { /** * Set a name for this task. * - * @param name task name Returns self + * @param name task name + * @return self * @see CallOptions.Builder#setName(java.lang.String) */ public T setName(String name) { @@ -27,7 +28,8 @@ public class BaseTaskCaller> { * times. If the same resource is set multiple times, the latest quantity will be used. * * @param name resource name - * @param value resource capacity Returns self + * @param value resource capacity + * @return self * @see CallOptions.Builder#setResource(java.lang.String, java.lang.Double) */ public T setResource(String name, Double value) { @@ -39,7 +41,8 @@ public class BaseTaskCaller> { * Set custom requirements for multiple resources. This method can be called multiple times. If * the same resource is set multiple times, the latest quantity will be used. * - * @param resources requirements for multiple resources. Returns self + * @param resources requirements for multiple resources. + * @return self * @see CallOptions.Builder#setResources(java.util.Map) */ public T setResources(Map resources) { diff --git a/java/api/src/main/java/io/ray/api/call/PyActorCreator.java b/java/api/src/main/java/io/ray/api/call/PyActorCreator.java index 5add65346..fb87a1eac 100644 --- a/java/api/src/main/java/io/ray/api/call/PyActorCreator.java +++ b/java/api/src/main/java/io/ray/api/call/PyActorCreator.java @@ -17,7 +17,7 @@ public class PyActorCreator extends BaseActorCreator { /** * Create a python actor remotely and return a handle to the created actor. * - *

Returns a handle to the created python actor. + * @return a handle to the created python actor. */ public PyActorHandle remote() { return Ray.internal().createActor(pyActorClass, args, buildOptions()); diff --git a/java/api/src/main/java/io/ray/api/call/PyActorTaskCaller.java b/java/api/src/main/java/io/ray/api/call/PyActorTaskCaller.java index c9444548f..7ee7d8a13 100644 --- a/java/api/src/main/java/io/ray/api/call/PyActorTaskCaller.java +++ b/java/api/src/main/java/io/ray/api/call/PyActorTaskCaller.java @@ -25,7 +25,7 @@ public class PyActorTaskCaller { * Execute a python actor method remotely and return an object reference to the result object in * the object store. * - *

Returns an object reference to an object in the object store. + * @return an object reference to an object in the object store. */ @SuppressWarnings("unchecked") public ObjectRef remote() { diff --git a/java/api/src/main/java/io/ray/api/call/PyTaskCaller.java b/java/api/src/main/java/io/ray/api/call/PyTaskCaller.java index 8d58e9b30..ecd7aa3c8 100644 --- a/java/api/src/main/java/io/ray/api/call/PyTaskCaller.java +++ b/java/api/src/main/java/io/ray/api/call/PyTaskCaller.java @@ -22,7 +22,7 @@ public class PyTaskCaller extends BaseTaskCaller> { * Execute a python function remotely and return an object reference to the result object in the * object store. * - *

Returns an object reference to an object in the object store. + * @return an object reference to an object in the object store. */ @SuppressWarnings("unchecked") public ObjectRef remote() { diff --git a/java/api/src/main/java/io/ray/api/call/TaskCaller.java b/java/api/src/main/java/io/ray/api/call/TaskCaller.java index 82f72d63e..80dacec2d 100644 --- a/java/api/src/main/java/io/ray/api/call/TaskCaller.java +++ b/java/api/src/main/java/io/ray/api/call/TaskCaller.java @@ -22,7 +22,7 @@ public class TaskCaller extends BaseTaskCaller> { * Execute a java function remotely and return an object reference to the result object in the * object store. * - *

Returns an object reference to an object in the object store. + * @return an object reference to an object in the object store. */ @SuppressWarnings("unchecked") public ObjectRef remote() { diff --git a/java/api/src/main/java/io/ray/api/function/PyActorClass.java b/java/api/src/main/java/io/ray/api/function/PyActorClass.java index c753e1f27..d76385919 100644 --- a/java/api/src/main/java/io/ray/api/function/PyActorClass.java +++ b/java/api/src/main/java/io/ray/api/function/PyActorClass.java @@ -38,7 +38,8 @@ public class PyActorClass { * Create a python actor class. * * @param moduleName The full module name of this actor class - * @param className The name of this actor class Returns a python actor class + * @param className The name of this actor class + * @return a python actor class */ public static PyActorClass of(String moduleName, String className) { return new PyActorClass(moduleName, className); diff --git a/java/api/src/main/java/io/ray/api/function/PyActorMethod.java b/java/api/src/main/java/io/ray/api/function/PyActorMethod.java index f91b0c9f9..6f24b5d11 100644 --- a/java/api/src/main/java/io/ray/api/function/PyActorMethod.java +++ b/java/api/src/main/java/io/ray/api/function/PyActorMethod.java @@ -43,7 +43,8 @@ public class PyActorMethod { /** * Create a python actor method. * - * @param methodName The name of this actor method Returns a python actor method. + * @param methodName The name of this actor method + * @return a python actor method. */ public static PyActorMethod of(String methodName) { return of(methodName, Object.class); @@ -54,7 +55,8 @@ public class PyActorMethod { * * @param methodName The name of this actor method * @param returnType Class of the return value of this actor method - * @param The type of the return value of this actor method Returns a python actor method. + * @param The type of the return value of this actor method + * @return a python actor method. */ public static PyActorMethod of(String methodName, Class returnType) { return new PyActorMethod<>(methodName, returnType); diff --git a/java/api/src/main/java/io/ray/api/function/PyFunction.java b/java/api/src/main/java/io/ray/api/function/PyFunction.java index 119bba4e5..2119b0bbf 100644 --- a/java/api/src/main/java/io/ray/api/function/PyFunction.java +++ b/java/api/src/main/java/io/ray/api/function/PyFunction.java @@ -49,7 +49,8 @@ public class PyFunction { * Create a python function. * * @param moduleName The full module name of this function - * @param functionName The name of this function Returns a python function. + * @param functionName The name of this function + * @return a python function. */ public static PyFunction of(String moduleName, String functionName) { return of(moduleName, functionName, Object.class); @@ -61,7 +62,8 @@ public class PyFunction { * @param moduleName The full module name of this function * @param functionName The name of this function * @param returnType Class of the return value of this function - * @param Type of the return value of this function Returns a python function. + * @param Type of the return value of this function + * @return a python function. */ public static PyFunction of(String moduleName, String functionName, Class returnType) { return new PyFunction<>(moduleName, functionName, returnType); diff --git a/java/api/src/main/java/io/ray/api/id/BaseId.java b/java/api/src/main/java/io/ray/api/id/BaseId.java index 573f549b2..ee91a77d6 100644 --- a/java/api/src/main/java/io/ray/api/id/BaseId.java +++ b/java/api/src/main/java/io/ray/api/id/BaseId.java @@ -52,7 +52,7 @@ public abstract class BaseId implements Serializable { /** * Derived class should implement this function. * - *

Returns The length of this id in bytes. + * @return The length of this id in bytes. */ public abstract int size(); diff --git a/java/api/src/main/java/io/ray/api/options/ActorCreationOptions.java b/java/api/src/main/java/io/ray/api/options/ActorCreationOptions.java index 29a13c115..303239735 100644 --- a/java/api/src/main/java/io/ray/api/options/ActorCreationOptions.java +++ b/java/api/src/main/java/io/ray/api/options/ActorCreationOptions.java @@ -50,7 +50,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * this name via {@link Ray#getActor(java.lang.String)}. If you want create a named actor that * is accessible from all jobs, use {@link Builder#setGlobalName(java.lang.String)} instead. * - * @param name The name of the named actor. Returns self + * @param name The name of the named actor. + * @return self */ public Builder setName(String name) { this.name = name; @@ -63,7 +64,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * {@link Ray#getGlobalActor(java.lang.String)}. If you want to create a named actor that is * only accessible from this job, use {@link Builder#setName(java.lang.String)} instead. * - * @param name The name of the named actor. Returns self + * @param name The name of the named actor. + * @return self */ public Builder setGlobalName(String name) { this.name = name; @@ -77,7 +79,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * will be used. * * @param resourceName resource name - * @param resourceQuantity resource quantity Returns self + * @param resourceQuantity resource quantity + * @return self */ public Builder setResource(String resourceName, Double resourceQuantity) { this.resources.put(resourceName, resourceQuantity); @@ -89,7 +92,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * be called multiple times. If the same resource is set multiple times, the latest quantity * will be used. * - * @param resources requirements for multiple resources. Returns self + * @param resources requirements for multiple resources. + * @return self */ public Builder setResources(Map resources) { this.resources.putAll(resources); @@ -101,7 +105,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * unexpectedly. The minimum valid value is 0 (default), which indicates that the actor doesn't * need to be restarted. A value of -1 indicates that an actor should be restarted indefinitely. * - * @param maxRestarts max number of actor restarts Returns self + * @param maxRestarts max number of actor restarts + * @return self */ public Builder setMaxRestarts(int maxRestarts) { this.maxRestarts = maxRestarts; @@ -113,7 +118,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * *

Note, if this is set, this actor won't share Java worker with other actors or tasks. * - * @param jvmOptions JVM options for the Java worker that this actor is running in. Returns self + * @param jvmOptions JVM options for the Java worker that this actor is running in. + * @return self */ public Builder setJvmOptions(String jvmOptions) { this.jvmOptions = jvmOptions; @@ -126,8 +132,8 @@ public class ActorCreationOptions extends BaseTaskOptions { *

The max concurrency defaults to 1 for threaded execution. Note that the execution order is * not guaranteed when {@code max_concurrency > 1}. * - * @param maxConcurrency The max number of concurrent calls to allow for this actor. Returns - * self + * @param maxConcurrency The max number of concurrent calls to allow for this actor. + * @return self */ public Builder setMaxConcurrency(int maxConcurrency) { if (maxConcurrency <= 0) { @@ -142,7 +148,8 @@ public class ActorCreationOptions extends BaseTaskOptions { * Set the placement group to place this actor in. * * @param group The placement group of the actor. - * @param bundleIndex The index of the bundle to place this actor in. Returns self + * @param bundleIndex The index of the bundle to place this actor in. + * @return self */ public Builder setPlacementGroup(PlacementGroup group, int bundleIndex) { this.group = group; diff --git a/java/api/src/main/java/io/ray/api/options/CallOptions.java b/java/api/src/main/java/io/ray/api/options/CallOptions.java index 233c30aa3..37e474d55 100644 --- a/java/api/src/main/java/io/ray/api/options/CallOptions.java +++ b/java/api/src/main/java/io/ray/api/options/CallOptions.java @@ -22,7 +22,8 @@ public class CallOptions extends BaseTaskOptions { /** * Set a name for this task. * - * @param name task name Returns self + * @param name task name + * @return self */ public Builder setName(String name) { this.name = name; @@ -34,7 +35,8 @@ public class CallOptions extends BaseTaskOptions { * multiple times. If the same resource is set multiple times, the latest quantity will be used. * * @param name resource name - * @param value resource capacity Returns self + * @param value resource capacity + * @return self */ public Builder setResource(String name, Double value) { this.resources.put(name, value); @@ -45,7 +47,8 @@ public class CallOptions extends BaseTaskOptions { * Set custom requirements for multiple resources. This method can be called multiple times. If * the same resource is set multiple times, the latest quantity will be used. * - * @param resources requirements for multiple resources. Returns self + * @param resources requirements for multiple resources. + * @return self */ public Builder setResources(Map resources) { this.resources.putAll(resources); diff --git a/java/api/src/main/java/io/ray/api/runtime/RayRuntime.java b/java/api/src/main/java/io/ray/api/runtime/RayRuntime.java index 2f3eeb2a7..53da3d48d 100644 --- a/java/api/src/main/java/io/ray/api/runtime/RayRuntime.java +++ b/java/api/src/main/java/io/ray/api/runtime/RayRuntime.java @@ -31,22 +31,24 @@ public interface RayRuntime { /** * Store an object in the object store. * - * @param obj The Java object to be stored. Returns A ObjectRef instance that represents the - * in-store object. + * @param obj The Java object to be stored. + * @return A ObjectRef instance that represents the in-store object. */ ObjectRef put(T obj); /** * Get an object from the object store. * - * @param objectRef The reference of the object to get. Returns The Java object. + * @param objectRef The reference of the object to get. + * @return The Java object. */ T get(ObjectRef objectRef); /** * Get a list of objects from the object store. * - * @param objectRefs The list of object references. Returns A list of Java objects. + * @param objectRefs The list of object references. + * @return A list of Java objects. */ List get(List> objectRefs); @@ -56,8 +58,8 @@ public interface RayRuntime { * * @param waitList A list of ObjectRef to wait for. * @param numReturns The number of objects that should be returned. - * @param timeoutMs The maximum time in milliseconds to wait before returning. Returns Two lists, - * one containing locally available objects, one containing the rest. + * @param timeoutMs The maximum time in milliseconds to wait before returning. + * @return Two lists, one containing locally available objects, one containing the rest. */ WaitResult wait(List> waitList, int numReturns, int timeoutMs); @@ -87,7 +89,8 @@ public interface RayRuntime { * name specified. * * @param name The name of the named actor. - * @param global Whether the named actor is global. Returns ActorHandle to the actor. + * @param global Whether the named actor is global. + * @return ActorHandle to the actor. */ Optional getActor(String name, boolean global); @@ -104,7 +107,8 @@ public interface RayRuntime { * * @param func The remote function to run. * @param args The arguments of the remote function. - * @param options The options for this call. Returns The result object. + * @param options The options for this call. + * @return The result object. */ ObjectRef call(RayFunc func, Object[] args, CallOptions options); @@ -113,7 +117,8 @@ public interface RayRuntime { * * @param pyFunction The Python function. * @param args Arguments of the function. - * @param options The options for this call. Returns The result object. + * @param options The options for this call. + * @return The result object. */ ObjectRef call(PyFunction pyFunction, Object[] args, CallOptions options); @@ -122,7 +127,8 @@ public interface RayRuntime { * * @param actor A handle to the actor. * @param func The remote function to run, it must be a method of the given actor. - * @param args The arguments of the remote function. Returns The result object. + * @param args The arguments of the remote function. + * @return The result object. */ ObjectRef callActor(ActorHandle actor, RayFunc func, Object[] args); @@ -131,7 +137,8 @@ public interface RayRuntime { * * @param pyActor A handle to the actor. * @param pyActorMethod The actor method. - * @param args Arguments of the function. Returns The result object. + * @param args Arguments of the function. + * @return The result object. */ ObjectRef callActor(PyActorHandle pyActor, PyActorMethod pyActorMethod, Object[] args); @@ -141,7 +148,8 @@ public interface RayRuntime { * @param actorFactoryFunc A remote function whose return value is the actor object. * @param args The arguments for the remote function. * @param The type of the actor object. - * @param options The options for creating actor. Returns A handle to the actor. + * @param options The options for creating actor. + * @return A handle to the actor. */ ActorHandle createActor( RayFunc actorFactoryFunc, Object[] args, ActorCreationOptions options); @@ -151,7 +159,8 @@ public interface RayRuntime { * * @param pyActorClass The Python actor class. * @param args Arguments of the actor constructor. - * @param options The options for creating actor. Returns A handle to the actor. + * @param options The options for creating actor. + * @return A handle to the actor. */ PyActorHandle createActor(PyActorClass pyActorClass, Object[] args, ActorCreationOptions options); @@ -170,14 +179,16 @@ public interface RayRuntime { /** * Wrap a {@link Runnable} with necessary context capture. * - * @param runnable The runnable to wrap. Returns The wrapped runnable. + * @param runnable The runnable to wrap. + * @return The wrapped runnable. */ Runnable wrapRunnable(Runnable runnable); /** * Wrap a {@link Callable} with necessary context capture. * - * @param callable The callable to wrap. Returns The wrapped callable. + * @param callable The callable to wrap. + * @return The wrapped callable. */ Callable wrapCallable(Callable callable); @@ -187,14 +198,15 @@ public interface RayRuntime { /** * Get a placement group by id. * - * @param id placement group id. Returns The placement group. + * @param id placement group id. + * @return The placement group. */ PlacementGroup getPlacementGroup(PlacementGroupId id); /** * Get all placement groups in this cluster. * - *

Returns All placement groups. + * @return All placement groups. */ List getAllPlacementGroups(); @@ -209,8 +221,8 @@ public interface RayRuntime { * Wait for the placement group to be ready within the specified time. * * @param id Id of placement group. - * @param timeoutMs Timeout in milliseconds. Returns True if the placement group is created. False - * otherwise. + * @param timeoutMs Timeout in milliseconds. + * @return True if the placement group is created. False otherwise. */ boolean waitPlacementGroupReady(PlacementGroupId id, int timeoutMs); } diff --git a/java/api/src/main/java/io/ray/api/runtimecontext/RuntimeContext.java b/java/api/src/main/java/io/ray/api/runtimecontext/RuntimeContext.java index b5fa486aa..d00ea4f11 100644 --- a/java/api/src/main/java/io/ray/api/runtimecontext/RuntimeContext.java +++ b/java/api/src/main/java/io/ray/api/runtimecontext/RuntimeContext.java @@ -21,7 +21,7 @@ public interface RuntimeContext { boolean wasCurrentActorRestarted(); /** - * Return true if Ray is running in single-process mode, false if Ray is running in cluster mode. + * Returns true if Ray is running in single-process mode, false if Ray is running in cluster mode. */ boolean isSingleProcess(); diff --git a/java/runtime/src/main/java/io/ray/runtime/actor/NativeActorHandle.java b/java/runtime/src/main/java/io/ray/runtime/actor/NativeActorHandle.java index 1dd4b84f5..85a46ad8b 100644 --- a/java/runtime/src/main/java/io/ray/runtime/actor/NativeActorHandle.java +++ b/java/runtime/src/main/java/io/ray/runtime/actor/NativeActorHandle.java @@ -71,7 +71,7 @@ public abstract class NativeActorHandle implements BaseActorHandle, Externalizab /** * Serialize this actor handle to bytes. * - *

Returns the bytes of the actor handle + * @return the bytes of the actor handle */ public byte[] toBytes() { return nativeSerialize(actorId); @@ -80,7 +80,7 @@ public abstract class NativeActorHandle implements BaseActorHandle, Externalizab /** * Deserialize an actor handle from bytes. * - *

Returns the bytes of an actor handle + * @return the bytes of an actor handle */ public static NativeActorHandle fromBytes(byte[] bytes) { byte[] actorId = nativeDeserialize(bytes); diff --git a/java/runtime/src/main/java/io/ray/runtime/functionmanager/FunctionManager.java b/java/runtime/src/main/java/io/ray/runtime/functionmanager/FunctionManager.java index d26a13dca..c9ef7ce3b 100644 --- a/java/runtime/src/main/java/io/ray/runtime/functionmanager/FunctionManager.java +++ b/java/runtime/src/main/java/io/ray/runtime/functionmanager/FunctionManager.java @@ -69,7 +69,8 @@ public class FunctionManager { * Get the RayFunction from a RayFunc instance (a lambda). * * @param jobId current job id. - * @param func The lambda. Returns A RayFunction object. + * @param func The lambda. + * @return A RayFunction object. */ public RayFunction getFunction(JobId jobId, RayFunc func) { JavaFunctionDescriptor functionDescriptor = RAY_FUNC_CACHE.get().get(func.getClass()); @@ -90,7 +91,8 @@ public class FunctionManager { * Get the RayFunction from a function descriptor. * * @param jobId Current job id. - * @param functionDescriptor The function descriptor. Returns A RayFunction object. + * @param functionDescriptor The function descriptor. + * @return A RayFunction object. */ public RayFunction getFunction(JobId jobId, JavaFunctionDescriptor functionDescriptor) { JobFunctionTable jobFunctionTable = jobFunctionTables.get(jobId); diff --git a/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java b/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java index df34212e7..cc70bbd7e 100644 --- a/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java +++ b/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java @@ -35,7 +35,8 @@ public class GcsClient { /** * Get placement group by {@link PlacementGroupId}. * - * @param placementGroupId Id of placement group. Returns The placement group. + * @param placementGroupId Id of placement group. + * @return The placement group. */ public PlacementGroup getPlacementGroupInfo(PlacementGroupId placementGroupId) { byte[] result = globalStateAccessor.getPlacementGroupInfo(placementGroupId); @@ -45,7 +46,7 @@ public class GcsClient { /** * Get all placement groups in this cluster. * - *

Returns All placement groups. + * @return All placement groups. */ public List getAllPlacementGroupInfo() { List results = globalStateAccessor.getAllPlacementGroupInfo(); diff --git a/java/runtime/src/main/java/io/ray/runtime/gcs/RedisClient.java b/java/runtime/src/main/java/io/ray/runtime/gcs/RedisClient.java index 77004a849..811402994 100644 --- a/java/runtime/src/main/java/io/ray/runtime/gcs/RedisClient.java +++ b/java/runtime/src/main/java/io/ray/runtime/gcs/RedisClient.java @@ -88,7 +88,7 @@ public class RedisClient { /** * Return the specified elements of the list stored at the specified key. * - *

Returns Multi bulk reply, specifically a list of elements in the specified range. + * @return Multi bulk reply, specifically a list of elements in the specified range. */ public List lrange(byte[] key, long start, long end) { try (Jedis jedis = jedisPool.getResource()) { diff --git a/java/runtime/src/main/java/io/ray/runtime/metric/Metric.java b/java/runtime/src/main/java/io/ray/runtime/metric/Metric.java index 961cbfe9a..80c39cf96 100644 --- a/java/runtime/src/main/java/io/ray/runtime/metric/Metric.java +++ b/java/runtime/src/main/java/io/ray/runtime/metric/Metric.java @@ -54,7 +54,7 @@ public abstract class Metric { /** * Get the value to record and then reset. * - *

Returns latest updating value. + * @return latest updating value. */ protected abstract double getAndReset(); diff --git a/java/runtime/src/main/java/io/ray/runtime/metric/Metrics.java b/java/runtime/src/main/java/io/ray/runtime/metric/Metrics.java index 85939ed79..f3af834f6 100644 --- a/java/runtime/src/main/java/io/ray/runtime/metric/Metrics.java +++ b/java/runtime/src/main/java/io/ray/runtime/metric/Metrics.java @@ -111,7 +111,7 @@ public final class Metrics { /** * Creates a metric by sub-class. * - *

Returns a metric + * @return a metric */ protected abstract M create(); diff --git a/java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java b/java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java index 76576b969..51ae9bfd2 100644 --- a/java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java +++ b/java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java @@ -55,7 +55,8 @@ public class ObjectSerializer { * Deserialize an object from an {@link NativeRayObject} instance. * * @param nativeRayObject The object to deserialize. - * @param objectId The associated object ID of the object. Returns The deserialized object. + * @param objectId The associated object ID of the object. + * @return The deserialized object. */ public static Object deserialize( NativeRayObject nativeRayObject, ObjectId objectId, Class objectType) { @@ -110,7 +111,8 @@ public class ObjectSerializer { /** * Serialize an Java object to an {@link NativeRayObject} instance. * - * @param object The object to serialize. Returns The serialized object. + * @param object The object to serialize. + * @return The serialized object. */ public static NativeRayObject serialize(Object object) { if (object instanceof NativeRayObject) { diff --git a/java/runtime/src/main/java/io/ray/runtime/object/ObjectStore.java b/java/runtime/src/main/java/io/ray/runtime/object/ObjectStore.java index df524af11..8711811b2 100644 --- a/java/runtime/src/main/java/io/ray/runtime/object/ObjectStore.java +++ b/java/runtime/src/main/java/io/ray/runtime/object/ObjectStore.java @@ -26,7 +26,8 @@ public abstract class ObjectStore { /** * Put a raw object into object store. * - * @param obj The ray object. Returns Generated ID of the object. + * @param obj The ray object. + * @return Generated ID of the object. */ public abstract ObjectId putRaw(NativeRayObject obj); @@ -41,7 +42,8 @@ public abstract class ObjectStore { /** * Serialize and put an object to the object store. * - * @param object The object to put. Returns Id of the object. + * @param object The object to put. + * @return Id of the object. */ public ObjectId put(Object object) { if (object instanceof NativeRayObject) { @@ -71,8 +73,8 @@ public abstract class ObjectStore { * Get a list of raw objects from the object store. * * @param objectIds IDs of the objects to get. - * @param timeoutMs Timeout in milliseconds, wait infinitely if it's negative. Returns Result list - * of objects data. + * @param timeoutMs Timeout in milliseconds, wait infinitely if it's negative. + * @return Result list of objects data. */ public abstract List getRaw(List objectIds, long timeoutMs); @@ -80,7 +82,8 @@ public abstract class ObjectStore { * Get a list of objects from the object store. * * @param ids List of the object ids. - * @param Type of these objects. Returns A list of GetResult objects. + * @param Type of these objects. + * @return A list of GetResult objects. */ @SuppressWarnings("unchecked") public List get(List ids, Class elementType) { @@ -118,8 +121,8 @@ public abstract class ObjectStore { * * @param objectIds IDs of the objects to wait for. * @param numObjects Number of objects that should appear. - * @param timeoutMs Timeout in milliseconds, wait infinitely if it's negative. Returns A bitset - * that indicates each object has appeared or not. + * @param timeoutMs Timeout in milliseconds, wait infinitely if it's negative. + * @return A bitset that indicates each object has appeared or not. */ public abstract List wait(List objectIds, int numObjects, long timeoutMs); @@ -129,8 +132,8 @@ public abstract class ObjectStore { * * @param waitList A list of object references to wait for. * @param numReturns The number of objects that should be returned. - * @param timeoutMs The maximum time in milliseconds to wait before returning. Returns Two lists, - * one containing locally available objects, one containing the rest. + * @param timeoutMs The maximum time in milliseconds to wait before returning. + * @return Two lists, one containing locally available objects, one containing the rest. */ public WaitResult wait(List> waitList, int numReturns, int timeoutMs) { Preconditions.checkNotNull(waitList); @@ -185,7 +188,8 @@ public abstract class ObjectStore { /** * Promote the given object to the underlying object store, and get the ownership info. * - * @param objectId The ID of the object to promote Returns the serialized ownership address + * @param objectId The ID of the object to promote + * @return the serialized ownership address */ public abstract byte[] promoteAndGetOwnershipInfo(ObjectId objectId); diff --git a/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupImpl.java b/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupImpl.java index b08f7c9f5..1d0d54084 100644 --- a/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupImpl.java +++ b/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupImpl.java @@ -53,8 +53,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Wait for the placement group to be ready within the specified time. * - * @param timeoutSeconds Timeout in seconds. Returns True if the placement group is created. False - * otherwise. + * @param timeoutSeconds Timeout in seconds. + * @return True if the placement group is created. False otherwise. */ public boolean wait(int timeoutSeconds) { return Ray.internal().waitPlacementGroupReady(id, timeoutSeconds); @@ -71,7 +71,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Set the Id of the placement group. * - * @param id Id of the placement group. Returns self. + * @param id Id of the placement group. + * @return self. */ public Builder setId(PlacementGroupId id) { this.id = id; @@ -81,7 +82,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Set the name of the placement group. * - * @param name Name of the placement group. Returns self. + * @param name Name of the placement group. + * @return self. */ public Builder setName(String name) { this.name = name; @@ -91,7 +93,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Set the bundles of the placement group. * - * @param bundles the bundles of the placement group. Returns self. + * @param bundles the bundles of the placement group. + * @return self. */ public Builder setBundles(List> bundles) { this.bundles = bundles; @@ -101,7 +104,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Set the placement strategy of the placement group. * - * @param strategy the placement strategy of the placement group. Returns self. + * @param strategy the placement strategy of the placement group. + * @return self. */ public Builder setStrategy(PlacementStrategy strategy) { this.strategy = strategy; @@ -111,7 +115,8 @@ public class PlacementGroupImpl implements PlacementGroup { /** * Set the placement state of the placement group. * - * @param state the state of the placement group. Returns self. + * @param state the state of the placement group. + * @return self. */ public Builder setState(PlacementGroupState state) { this.state = state; diff --git a/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupUtils.java b/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupUtils.java index 75305ef1f..8e9d03cc6 100644 --- a/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupUtils.java +++ b/java/runtime/src/main/java/io/ray/runtime/placementgroup/PlacementGroupUtils.java @@ -61,8 +61,8 @@ public class PlacementGroupUtils { /** * Generate a PlacementGroupImpl from placementGroupTableData protobuf data. * - * @param placementGroupTableData protobuf data. Returns placement group info {@link - * PlacementGroupImpl} + * @param placementGroupTableData protobuf data. + * @return placement group info {@link PlacementGroupImpl} */ private static PlacementGroupImpl generatePlacementGroupFromPbData( PlacementGroupTableData placementGroupTableData) { @@ -90,8 +90,8 @@ public class PlacementGroupUtils { /** * Generate a PlacementGroupImpl from byte array. * - * @param placementGroupByteArray bytes array from native method. Returns placement group info - * {@link PlacementGroupImpl} + * @param placementGroupByteArray bytes array from native method. + * @return placement group info {@link PlacementGroupImpl} */ public static PlacementGroupImpl generatePlacementGroupFromByteArray( byte[] placementGroupByteArray) { diff --git a/java/runtime/src/main/java/io/ray/runtime/task/TaskSubmitter.java b/java/runtime/src/main/java/io/ray/runtime/task/TaskSubmitter.java index ca195d6ce..e8a835171 100644 --- a/java/runtime/src/main/java/io/ray/runtime/task/TaskSubmitter.java +++ b/java/runtime/src/main/java/io/ray/runtime/task/TaskSubmitter.java @@ -21,7 +21,8 @@ public interface TaskSubmitter { * @param functionDescriptor The remote function to execute. * @param args Arguments of this task. * @param numReturns Return object count. - * @param options Options for this task. Returns Ids of the return objects. + * @param options Options for this task. + * @return Ids of the return objects. */ List submitTask( FunctionDescriptor functionDescriptor, @@ -34,7 +35,8 @@ public interface TaskSubmitter { * * @param functionDescriptor The remote function that generates the actor object. * @param args Arguments of this task. - * @param options Options for this actor creation task. Returns Handle to the actor. + * @param options Options for this actor creation task. + * @return Handle to the actor. * @throws IllegalArgumentException if actor of specified name exists */ BaseActorHandle createActor( @@ -48,7 +50,8 @@ public interface TaskSubmitter { * @param functionDescriptor The remote function to execute. * @param args Arguments of this task. * @param numReturns Return object count. - * @param options Options for this task. Returns Ids of the return objects. + * @param options Options for this task. + * @return Ids of the return objects. */ List submitActorTask( BaseActorHandle actor, @@ -62,7 +65,8 @@ public interface TaskSubmitter { * * @param name Name of the placement group. * @param bundles Pre-allocated resource list. - * @param strategy Actor placement strategy. Returns A handle to the created placement group. + * @param strategy Actor placement strategy. + * @return A handle to the created placement group. */ PlacementGroup createPlacementGroup( String name, List> bundles, PlacementStrategy strategy); @@ -78,8 +82,8 @@ public interface TaskSubmitter { * Wait for the placement group to be ready within the specified time. * * @param id Id of placement group. - * @param timeoutMs Timeout in milliseconds. Returns True if the placement group is created. False - * otherwise. + * @param timeoutMs Timeout in milliseconds. + * @return True if the placement group is created. False otherwise. */ boolean waitPlacementGroupReady(PlacementGroupId id, int timeoutMs); diff --git a/java/runtime/src/main/java/io/ray/runtime/util/BinaryFileUtil.java b/java/runtime/src/main/java/io/ray/runtime/util/BinaryFileUtil.java index 85c327a44..f3282ed08 100644 --- a/java/runtime/src/main/java/io/ray/runtime/util/BinaryFileUtil.java +++ b/java/runtime/src/main/java/io/ray/runtime/util/BinaryFileUtil.java @@ -21,7 +21,8 @@ public class BinaryFileUtil { * will be protected by a file lock. * * @param destDir a directory to extract resource file to - * @param fileName resource file name Returns extracted resource file + * @param fileName resource file name + * @return extracted resource file */ public static File getNativeFile(String destDir, String fileName) { final File dir = new File(destDir); diff --git a/java/runtime/src/main/java/io/ray/runtime/util/IdUtil.java b/java/runtime/src/main/java/io/ray/runtime/util/IdUtil.java index 4f7bf2580..239568afa 100644 --- a/java/runtime/src/main/java/io/ray/runtime/util/IdUtil.java +++ b/java/runtime/src/main/java/io/ray/runtime/util/IdUtil.java @@ -13,7 +13,7 @@ public class IdUtil { /** * Compute the actor ID of the task which created this object. * - *

Returns The actor ID of the task which created this object. + * @return The actor ID of the task which created this object. */ public static ActorId getActorIdFromObjectId(ObjectId objectId) { byte[] taskIdBytes = new byte[TaskId.LENGTH]; diff --git a/java/runtime/src/main/java/io/ray/runtime/util/ResourceUtil.java b/java/runtime/src/main/java/io/ray/runtime/util/ResourceUtil.java index 0c7a93d27..e9676d07b 100644 --- a/java/runtime/src/main/java/io/ray/runtime/util/ResourceUtil.java +++ b/java/runtime/src/main/java/io/ray/runtime/util/ResourceUtil.java @@ -11,8 +11,8 @@ public class ResourceUtil { * Convert resources map to a string that is used for the command line argument of starting * raylet. * - * @param resources The resources map to be converted. Returns The starting-raylet command line - * argument, like "CPU,4,GPU,0". + * @param resources The resources map to be converted. + * @return The starting-raylet command line argument, like "CPU,4,GPU,0". */ public static String getResourcesStringFromMap(Map resources) { StringBuilder builder = new StringBuilder(); @@ -32,8 +32,9 @@ public class ResourceUtil { /** * Parse the static resources configure field and convert to the resources map. * - * @param resources The static resources string to be parsed. Returns The map whose key represents - * the resource name and the value represents the resource quantity. + * @param resources The static resources string to be parsed. + * @return The map whose key represents the resource name and the value represents the resource + * quantity. * @throws IllegalArgumentException If the resources string's format does match, it will throw an * IllegalArgumentException. */ diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/Function.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/Function.java index fbfc4736e..c12bdf87c 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/Function.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/Function.java @@ -11,7 +11,7 @@ public interface Function extends Serializable { * storage, and load it back when in fail-over through. {@link * Function#loadCheckpoint(Serializable)}. * - *

Returns A serializable object which represents function state. + * @return A serializable object which represents function state. */ default Serializable saveCheckpoint() { return null; diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/impl/FilterFunction.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/impl/FilterFunction.java index 877a93ae0..d60e335a9 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/impl/FilterFunction.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/function/impl/FilterFunction.java @@ -14,8 +14,8 @@ public interface FilterFunction extends Function { /** * The filter function that evaluates the predicate. * - * @param value The value to be filtered. Returns True for values that should be retained, false - * for values to be filtered out. + * @param value The value to be filtered. + * @return True for values that should be retained, false for values to be filtered out. */ boolean filter(T value) throws Exception; } diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/partition/Partition.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/partition/Partition.java index 527f469c3..80e9d9272 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/partition/Partition.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/partition/Partition.java @@ -15,8 +15,8 @@ public interface Partition extends Function { * record. * * @param record The record. - * @param numPartition num of partitions Returns IDs of the downstream partitions that should - * receive the record. + * @param numPartition num of partitions + * @return IDs of the downstream partitions that should receive the record. */ int[] partition(T record, int numPartition); } diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStream.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStream.java index 698eab29d..999057d5a 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStream.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStream.java @@ -59,7 +59,8 @@ public class DataStream extends Stream, T> { * Apply a map function to this stream. * * @param mapFunction The map function. - * @param Type of data returned by the map function. Returns A new DataStream. + * @param Type of data returned by the map function. + * @return A new DataStream. */ public DataStream map(MapFunction mapFunction) { return new DataStream<>(this, new MapOperator<>(mapFunction)); @@ -69,7 +70,8 @@ public class DataStream extends Stream, T> { * Apply a flat-map function to this stream. * * @param flatMapFunction The FlatMapFunction - * @param Type of data returned by the flatmap function. Returns A new DataStream + * @param Type of data returned by the flatmap function. + * @return A new DataStream */ public DataStream flatMap(FlatMapFunction flatMapFunction) { return new DataStream<>(this, new FlatMapOperator<>(flatMapFunction)); @@ -84,7 +86,8 @@ public class DataStream extends Stream, T> { * type with each other. * * @param stream The DataStream to union output with. - * @param others The other DataStreams to union output with. Returns A new UnionStream. + * @param others The other DataStreams to union output with. + * @return A new UnionStream. */ @SafeVarargs public final DataStream union(DataStream stream, DataStream... others) { @@ -98,7 +101,8 @@ public class DataStream extends Stream, T> { * Apply union transformations to this stream by merging {@link DataStream} outputs of the same * type with each other. * - * @param streams The DataStreams to union output with. Returns A new UnionStream. + * @param streams The DataStreams to union output with. + * @return A new UnionStream. */ public final DataStream union(List> streams) { if (this instanceof UnionStream) { @@ -115,7 +119,8 @@ public class DataStream extends Stream, T> { * * @param other Another stream. * @param The type of the other stream data. - * @param The type of the data in the joined stream. Returns A new JoinStream. + * @param The type of the data in the joined stream. + * @return A new JoinStream. */ public JoinStream join(DataStream other) { return new JoinStream<>(this, other); @@ -129,7 +134,8 @@ public class DataStream extends Stream, T> { /** * Apply a sink function and get a StreamSink. * - * @param sinkFunction The sink function. Returns A new StreamSink. + * @param sinkFunction The sink function. + * @return A new StreamSink. */ public DataStreamSink sink(SinkFunction sinkFunction) { return new DataStreamSink<>(this, new SinkOperator<>(sinkFunction)); @@ -139,7 +145,8 @@ public class DataStream extends Stream, T> { * Apply a key-by function to this stream. * * @param keyFunction the key function. - * @param The type of the key. Returns A new KeyDataStream. + * @param The type of the key. + * @return A new KeyDataStream. */ public KeyDataStream keyBy(KeyFunction keyFunction) { checkPartitionCall(); @@ -149,7 +156,7 @@ public class DataStream extends Stream, T> { /** * Apply broadcast to this stream. * - *

Returns This stream. + * @return This stream. */ public DataStream broadcast() { checkPartitionCall(); @@ -159,7 +166,8 @@ public class DataStream extends Stream, T> { /** * Apply a partition to this stream. * - * @param partition The partitioning strategy. Returns This stream. + * @param partition The partitioning strategy. + * @return This stream. */ public DataStream partitionBy(Partition partition) { checkPartitionCall(); diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStreamSource.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStreamSource.java index 13de0b33b..53dd2a097 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStreamSource.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/DataStreamSource.java @@ -27,7 +27,8 @@ public class DataStreamSource extends DataStream implements StreamSource The type of source data. Returns A DataStreamSource. + * @param The type of source data. + * @return A DataStreamSource. */ public static DataStreamSource fromCollection( StreamingContext context, Collection values) { diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/KeyDataStream.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/KeyDataStream.java index fb6431ef2..c50b23269 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/KeyDataStream.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/KeyDataStream.java @@ -33,7 +33,8 @@ public class KeyDataStream extends DataStream { /** * Apply a reduce function to this stream. * - * @param reduceFunction The reduce function. Returns A new DataStream. + * @param reduceFunction The reduce function. + * @return A new DataStream. */ public DataStream reduce(ReduceFunction reduceFunction) { return new DataStream<>(this, new ReduceOperator(reduceFunction)); @@ -44,7 +45,8 @@ public class KeyDataStream extends DataStream { * * @param aggregateFunction The aggregate function * @param The type of aggregated intermediate data. - * @param The type of result data. Returns A new DataStream. + * @param The type of result data. + * @return A new DataStream. */ public DataStream aggregate(AggregateFunction aggregateFunction) { return new DataStream<>(this, null); diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/jobgraph/JobGraph.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/jobgraph/JobGraph.java index 6e40ee441..b192dbcc8 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/jobgraph/JobGraph.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/jobgraph/JobGraph.java @@ -43,7 +43,7 @@ public class JobGraph implements Serializable { * Generate direct-graph(made up of a set of vertices and connected by edges) by current job graph * for simple log printing. * - *

Returns Digraph in string type. + * @return Digraph in string type. */ public String generateDigraph() { StringBuilder digraph = new StringBuilder(); diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonDataStream.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonDataStream.java index 25b587310..90f018ecd 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonDataStream.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonDataStream.java @@ -51,7 +51,8 @@ public class PythonDataStream extends Stream implement /** * Apply a map function to this stream. * - * @param func The python MapFunction. Returns A new PythonDataStream. + * @param func The python MapFunction. + * @return A new PythonDataStream. */ public PythonDataStream map(PythonFunction func) { func.setFunctionInterface(FunctionInterface.MAP_FUNCTION); @@ -65,7 +66,8 @@ public class PythonDataStream extends Stream implement /** * Apply a flat-map function to this stream. * - * @param func The python FlapMapFunction. Returns A new PythonDataStream + * @param func The python FlapMapFunction. + * @return A new PythonDataStream */ public PythonDataStream flatMap(PythonFunction func) { func.setFunctionInterface(FunctionInterface.FLAT_MAP_FUNCTION); @@ -79,8 +81,9 @@ public class PythonDataStream extends Stream implement /** * Apply a filter function to this stream. * - * @param func The python FilterFunction. Returns A new PythonDataStream that contains only the - * elements satisfying the given filter predicate. + * @param func The python FilterFunction. + * @return A new PythonDataStream that contains only the elements satisfying the given filter + * predicate. */ public PythonDataStream filter(PythonFunction func) { func.setFunctionInterface(FunctionInterface.FILTER_FUNCTION); @@ -92,7 +95,8 @@ public class PythonDataStream extends Stream implement * same type with each other. * * @param stream The DataStream to union output with. - * @param others The other DataStreams to union output with. Returns A new UnionStream. + * @param others The other DataStreams to union output with. + * @return A new UnionStream. */ public final PythonDataStream union(PythonDataStream stream, PythonDataStream... others) { List streams = new ArrayList<>(); @@ -105,7 +109,8 @@ public class PythonDataStream extends Stream implement * Apply union transformations to this stream by merging {@link PythonDataStream} outputs of the * same type with each other. * - * @param streams The DataStreams to union output with. Returns A new UnionStream. + * @param streams The DataStreams to union output with. + * @return A new UnionStream. */ public final PythonDataStream union(List streams) { if (this instanceof PythonUnionStream) { @@ -124,7 +129,8 @@ public class PythonDataStream extends Stream implement /** * Apply a sink function and get a StreamSink. * - * @param func The python SinkFunction. Returns A new StreamSink. + * @param func The python SinkFunction. + * @return A new StreamSink. */ public PythonStreamSink sink(PythonFunction func) { func.setFunctionInterface(FunctionInterface.SINK_FUNCTION); @@ -138,7 +144,8 @@ public class PythonDataStream extends Stream implement /** * Apply a key-by function to this stream. * - * @param func the python keyFunction. Returns A new KeyDataStream. + * @param func the python keyFunction. + * @return A new KeyDataStream. */ public PythonKeyDataStream keyBy(PythonFunction func) { checkPartitionCall(); @@ -149,7 +156,7 @@ public class PythonDataStream extends Stream implement /** * Apply broadcast to this stream. * - *

Returns This stream. + * @return This stream. */ public PythonDataStream broadcast() { checkPartitionCall(); @@ -159,7 +166,8 @@ public class PythonDataStream extends Stream implement /** * Apply a partition to this stream. * - * @param partition The partitioning strategy. Returns This stream. + * @param partition The partitioning strategy. + * @return This stream. */ public PythonDataStream partitionBy(PythonPartition partition) { checkPartitionCall(); diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonKeyDataStream.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonKeyDataStream.java index 8116fd392..078f84ac4 100644 --- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonKeyDataStream.java +++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/stream/PythonKeyDataStream.java @@ -31,7 +31,8 @@ public class PythonKeyDataStream extends PythonDataStream implements PythonStrea /** * Apply a reduce function to this stream. * - * @param func The reduce function. Returns A new DataStream. + * @param func The reduce function. + * @return A new DataStream. */ public PythonDataStream reduce(PythonFunction func) { func.setFunctionInterface(FunctionInterface.REDUCE_FUNCTION); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/global/CommonConfig.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/global/CommonConfig.java index 0c555e7c5..2ec3b6dfb 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/global/CommonConfig.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/global/CommonConfig.java @@ -11,7 +11,7 @@ public interface CommonConfig extends Config { /** * Ray streaming job id. Non-custom. * - *

Returns Job id with string type. + * @return Job id with string type. */ @DefaultValue(value = "default-job-id") @Key(value = JOB_ID) @@ -20,7 +20,7 @@ public interface CommonConfig extends Config { /** * Ray streaming job name. Non-custom. * - *

Returns Job name with string type. + * @return Job name with string type. */ @DefaultValue(value = "default-job-name") @Key(value = JOB_NAME) diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/master/SchedulerConfig.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/master/SchedulerConfig.java index bc2fc2bd3..79189431a 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/master/SchedulerConfig.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/config/master/SchedulerConfig.java @@ -11,7 +11,7 @@ public interface SchedulerConfig extends Config { /** * The timeout ms of worker initiation. Default is: 10000ms(10s). * - *

Returns timeout ms + * @return timeout ms */ @Key(WORKER_INITIATION_WAIT_TIMEOUT_MS) @DefaultValue(value = "10000") @@ -20,7 +20,7 @@ public interface SchedulerConfig extends Config { /** * The timeout ms of worker starting. Default is: 10000ms(10s). * - *

Returns timeout ms + * @return timeout ms */ @Key(WORKER_STARTING_WAIT_TIMEOUT_MS) @DefaultValue(value = "10000") diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/context/ContextBackend.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/context/ContextBackend.java index faf870390..83b62696e 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/context/ContextBackend.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/context/ContextBackend.java @@ -12,14 +12,15 @@ public interface ContextBackend { /** * check if key exists in state * - *

Returns true if exists + * @return true if exists */ boolean exists(final String key) throws Exception; /** * get content by key * - * @param key key Returns the StateBackend + * @param key key + * @return the StateBackend */ byte[] get(final String key) throws Exception; diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java index b0d3b522e..2852e0f99 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionGraph.java @@ -156,7 +156,7 @@ public class ExecutionGraph implements Serializable { /** * Get all execution vertices from current execution graph. * - *

Returns all execution vertices. + * @return all execution vertices. */ public List getAllExecutionVertices() { return executionJobVertexMap.values().stream() @@ -168,7 +168,7 @@ public class ExecutionGraph implements Serializable { /** * Get all execution vertices whose status is 'TO_ADD' from current execution graph. * - *

Returns all added execution vertices. + * @return all added execution vertices. */ public List getAllAddedExecutionVertices() { return executionJobVertexMap.values().stream() @@ -181,7 +181,8 @@ public class ExecutionGraph implements Serializable { /** * Get specified execution vertex from current execution graph by execution vertex id. * - * @param executionVertexId execution vertex id. Returns the specified execution vertex. + * @param executionVertexId execution vertex id. + * @return the specified execution vertex. */ public ExecutionVertex getExecutionVertexByExecutionVertexId(int executionVertexId) { if (executionVertexMap.containsKey(executionVertexId)) { @@ -193,7 +194,8 @@ public class ExecutionGraph implements Serializable { /** * Get specified execution vertex from current execution graph by actor id. * - * @param actorId the actor id of execution vertex. Returns the specified execution vertex. + * @param actorId the actor id of execution vertex. + * @return the specified execution vertex. */ public ExecutionVertex getExecutionVertexByActorId(ActorId actorId) { return actorIdExecutionVertexMap.get(actorId); @@ -202,7 +204,8 @@ public class ExecutionGraph implements Serializable { /** * Get specified actor by actor id. * - * @param actorId the actor id of execution vertex. Returns the specified actor handle. + * @param actorId the actor id of execution vertex. + * @return the specified actor handle. */ public Optional getActorById(ActorId actorId) { return getAllActors().stream().filter(actor -> actor.getId().equals(actorId)).findFirst(); @@ -212,7 +215,8 @@ public class ExecutionGraph implements Serializable { * Get the peer actor in the other side of channelName of a given actor * * @param actor actor in this side - * @param channelName the channel name Returns the peer actor in the other side + * @param channelName the channel name + * @return the peer actor in the other side */ public BaseActorHandle getPeerActor(BaseActorHandle actor, String channelName) { Set set = getActorsByChannelId(channelName); @@ -229,7 +233,8 @@ public class ExecutionGraph implements Serializable { /** * Get actors in both sides of a channelId * - * @param channelId the channelId Returns actors in both sides + * @param channelId the channelId + * @return actors in both sides */ public Set getActorsByChannelId(String channelId) { return channelGroupedActors.getOrDefault(channelId, Sets.newHashSet()); @@ -238,7 +243,7 @@ public class ExecutionGraph implements Serializable { /** * Get all actors by graph. * - *

Returns actor list + * @return actor list */ public List getAllActors() { return getActorsFromJobVertices(getExecutionJobVertexList()); @@ -247,7 +252,7 @@ public class ExecutionGraph implements Serializable { /** * Get source actors by graph. * - *

Returns actor list + * @return actor list */ public List getSourceActors() { List executionJobVertices = @@ -261,7 +266,7 @@ public class ExecutionGraph implements Serializable { /** * Get transformation and sink actors by graph. * - *

Returns actor list + * @return actor list */ public List getNonSourceActors() { List executionJobVertices = @@ -278,7 +283,7 @@ public class ExecutionGraph implements Serializable { /** * Get sink actors by graph. * - *

Returns actor list + * @return actor list */ public List getSinkActors() { List executionJobVertices = @@ -292,7 +297,8 @@ public class ExecutionGraph implements Serializable { /** * Get actors according to job vertices. * - * @param executionJobVertices specified job vertices Returns actor list + * @param executionJobVertices specified job vertices + * @return actor list */ public List getActorsFromJobVertices( List executionJobVertices) { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionJobVertex.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionJobVertex.java index 0aa426672..cf869c0c4 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionJobVertex.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionJobVertex.java @@ -109,7 +109,7 @@ public class ExecutionJobVertex implements Serializable { /** * e.g. 1-SourceOperator * - *

Returns operator name with index + * @return operator name with index */ public String getExecutionJobVertexNameWithIndex() { return executionJobVertexId + "-" + executionJobVertexName; diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java index b0dec4aef..9b07d131f 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/resource/Resources.java @@ -24,7 +24,7 @@ public class Resources implements Serializable { /** * Get registered containers, the container list is read-only. * - *

Returns container list. + * @return container list. */ public ImmutableList getRegisteredContainers() { return ImmutableList.copyOf(registerContainers); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/JobMaster.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/JobMaster.java index a1dd5b6bc..fd672978a 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/JobMaster.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/JobMaster.java @@ -101,7 +101,7 @@ public class JobMaster { /** * Init JobMaster. To initiate or recover other components(like metrics and extra coordinators). * - *

Returns init result + * @return init result */ public Boolean init(boolean isRecover) { LOG.info("Initializing job master, isRecover={}.", isRecover); @@ -136,7 +136,8 @@ public class JobMaster { * * * @param jobMasterActor JobMaster actor - * @param jobGraph logical plan Returns submit result + * @param jobGraph logical plan + * @return submit result */ public boolean submitJob(ActorHandle jobMasterActor, JobGraph jobGraph) { LOG.info("Begin submitting job using logical plan: {}.", jobGraph); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/graphmanager/GraphManager.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/graphmanager/GraphManager.java index ce8dd4741..b563917d9 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/graphmanager/GraphManager.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/graphmanager/GraphManager.java @@ -19,21 +19,22 @@ public interface GraphManager { /** * Build execution graph from job graph. * - * @param jobGraph logical plan of streaming job. Returns physical plan of streaming job. + * @param jobGraph logical plan of streaming job. + * @return physical plan of streaming job. */ ExecutionGraph buildExecutionGraph(JobGraph jobGraph); /** * Get job graph. * - *

Returns the job graph. + * @return the job graph. */ JobGraph getJobGraph(); /** * Get execution graph. * - *

Returns the execution graph. + * @return the execution graph. */ ExecutionGraph getExecutionGraph(); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java index 43671eea1..fbe3f696a 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/ResourceManager.java @@ -10,7 +10,7 @@ public interface ResourceManager extends ResourceAssignStrategy { /** * Get registered containers, the container list is read-only. * - *

Returns the registered container list + * @return the registered container list */ ImmutableList getRegisteredContainers(); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java index 8df20790c..9ce131d25 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/ResourceAssignStrategy.java @@ -13,7 +13,8 @@ public interface ResourceAssignStrategy { * Assign {@link Container} for {@link ExecutionVertex} * * @param containers registered container - * @param executionGraph execution graph Returns allocating view + * @param executionGraph execution graph + * @return allocating view */ ResourceAssignmentView assignResource(List containers, ExecutionGraph executionGraph); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java index 74b646c67..48f2366cd 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/resourcemanager/strategy/impl/PipelineFirstStrategy.java @@ -42,8 +42,8 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy { * Assign resource to each execution vertex in the given execution graph. * * @param containers registered containers - * @param executionGraph execution graph Returns allocating map, key is container ID, value is - * list of vertextId, and contains vertices + * @param executionGraph execution graph + * @return allocating map, key is container ID, value is list of vertextId, and contains vertices */ @Override public ResourceAssignmentView assignResource( @@ -133,7 +133,8 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy { * Find a container which matches required resource * * @param requiredResource required resource - * @param containers registered containers Returns container that matches the required resource + * @param containers registered containers + * @return container that matches the required resource */ private Container findMatchedContainer( Map requiredResource, List containers) { @@ -159,7 +160,8 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy { * Check if current container has enough resource * * @param requiredResource required resource - * @param container container Returns true if matches, false else + * @param container container + * @return true if matches, false else */ private boolean hasEnoughResource(Map requiredResource, Container container) { LOG.info("Check resource for index: {}, container: {}", currentContainerIndex, container); @@ -200,7 +202,8 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy { /** * Forward to next container * - * @param containers registered container list Returns next container in the list + * @param containers registered container list + * @return next container in the list */ private Container forwardToNextContainer(List containers) { this.currentContainerIndex = (this.currentContainerIndex + 1) % containers.size(); @@ -210,7 +213,8 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy { /** * Get current container * - * @param containers registered container Returns current container to allocate actor + * @param containers registered container + * @return current container to allocate actor */ private Container getCurrentContainer(List containers) { return containers.get(currentContainerIndex); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobScheduler.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobScheduler.java index 962c0bdfa..d0fb60d54 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobScheduler.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobScheduler.java @@ -8,7 +8,8 @@ public interface JobScheduler { /** * Schedule streaming job using the physical plan. * - * @param executionGraph physical plan Returns scheduling result + * @param executionGraph physical plan + * @return scheduling result */ boolean scheduleJob(ExecutionGraph executionGraph); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobSchedulerImpl.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobSchedulerImpl.java index 6309bb334..039715ccb 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobSchedulerImpl.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/JobSchedulerImpl.java @@ -95,7 +95,8 @@ public class JobSchedulerImpl implements JobScheduler { /** * Create JobWorker actors according to the physical plan. * - * @param executionGraph physical plan Returns actor creation result + * @param executionGraph physical plan + * @return actor creation result */ public boolean createWorkers(ExecutionGraph executionGraph) { LOG.info("Begin creating workers."); @@ -148,7 +149,8 @@ public class JobSchedulerImpl implements JobScheduler { /** * Build workers context. * - * @param executionGraph execution graph Returns vertex to worker context map + * @param executionGraph execution graph + * @return vertex to worker context map */ protected Map buildWorkersContext( ExecutionGraph executionGraph) { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/controller/WorkerLifecycleController.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/controller/WorkerLifecycleController.java index f5c4be5f7..3cd3984b2 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/controller/WorkerLifecycleController.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/master/scheduler/controller/WorkerLifecycleController.java @@ -36,7 +36,8 @@ public class WorkerLifecycleController { /** * Create JobWorker actor according to the execution vertex. * - * @param executionVertex target execution vertex Returns creation result + * @param executionVertex target execution vertex + * @return creation result */ private boolean createWorker(ExecutionVertex executionVertex) { LOG.info( @@ -84,7 +85,8 @@ public class WorkerLifecycleController { * Using context to init JobWorker. * * @param vertexToContextMap target JobWorker actor - * @param timeout timeout for waiting, unit: ms Returns initiation result + * @param timeout timeout for waiting, unit: ms + * @return initiation result */ public boolean initWorkers( Map vertexToContextMap, int timeout) { @@ -120,7 +122,8 @@ public class WorkerLifecycleController { * Start JobWorkers to run task. * * @param executionGraph physical plan - * @param timeout timeout for waiting, unit: ms Returns starting result + * @param timeout timeout for waiting, unit: ms + * @return starting result */ public boolean startWorkers(ExecutionGraph executionGraph, long lastCheckpointId, int timeout) { LOG.info("Begin starting workers."); @@ -150,7 +153,8 @@ public class WorkerLifecycleController { /** * Stop and destroy JobWorkers' actor. * - * @param executionVertices target vertices Returns destroy result + * @param executionVertices target vertices + * @return destroy result */ public boolean destroyWorkers(List executionVertices) { return asyncBatchExecute(this::destroyWorker, executionVertices); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/rpc/RemoteCallWorker.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/rpc/RemoteCallWorker.java index 5a5475350..6cd788138 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/rpc/RemoteCallWorker.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/rpc/RemoteCallWorker.java @@ -25,7 +25,8 @@ public class RemoteCallWorker { * Call JobWorker actor to init. * * @param actor target JobWorker actor - * @param context JobWorker's context Returns init result + * @param context JobWorker's context + * @return init result */ public static ObjectRef initWorker(BaseActorHandle actor, JobWorkerContext context) { LOG.info("Call worker to initiate, actor: {}, context: {}.", actor.getId(), context); @@ -50,7 +51,8 @@ public class RemoteCallWorker { * Call JobWorker actor to start. * * @param actor target JobWorker actor - * @param checkpointId checkpoint ID to be rollback Returns start result + * @param checkpointId checkpoint ID to be rollback + * @return start result */ public static ObjectRef rollback(BaseActorHandle actor, final Long checkpointId) { LOG.info("Call worker to start, actor: {}.", actor.getId()); @@ -79,7 +81,8 @@ public class RemoteCallWorker { /** * Call JobWorker actor to destroy without reconstruction. * - * @param actor target JobWorker actor Returns destroy result + * @param actor target JobWorker actor + * @return destroy result */ public static Boolean shutdownWithoutReconstruction(BaseActorHandle actor) { LOG.info("Call worker to shutdown without reconstruction, actor is {}.", actor.getId()); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java index 17ab4fe1e..ff3c62fee 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java @@ -115,7 +115,8 @@ public class DataReader { /** * Read message from input channels, if timeout, return null. * - * @param timeoutMillis timeout Returns message or null + * @param timeoutMillis timeout + * @return message or null */ public ChannelMessage read(long timeoutMillis) { if (buf.isEmpty()) { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java index d3a4b8d71..731031d62 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java @@ -86,7 +86,8 @@ public class ChannelId { * Generate channel name, which will be {@link ChannelId#ID_LENGTH} character * * @param fromTaskId upstream task id - * @param toTaskId downstream task id Returns channel name + * @param toTaskId downstream task id + * @return channel name */ public static String genIdStr(int fromTaskId, int toTaskId, long ts) { /* @@ -116,7 +117,8 @@ public class ChannelId { } /** - * @param id hex string representation of channel id Returns bytes representation of channel id + * @param id hex string representation of channel id + * @return bytes representation of channel id */ public static byte[] idStrToBytes(String id) { byte[] idBytes = BaseEncoding.base16().decode(id.toUpperCase()); @@ -125,7 +127,8 @@ public class ChannelId { } /** - * @param id bytes representation of channel id Returns hex string representation of channel id + * @param id bytes representation of channel id + * @return hex string representation of channel id */ public static String idBytesToStr(byte[] id) { assert id.length == ChannelId.ID_LENGTH; diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/EnvUtil.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/EnvUtil.java index 07fda18a6..29ac29f4d 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/EnvUtil.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/EnvUtil.java @@ -36,7 +36,7 @@ public class EnvUtil { /** * Execute an external command. * - *

Returns Whether the command succeeded. + * @return Whether the command succeeded. */ public static boolean executeCommand(List command, int waitTimeoutSeconds) { try { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/Platform.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/Platform.java index effafcc54..324e1ab9d 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/Platform.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/Platform.java @@ -77,7 +77,10 @@ public final class Platform { buffer.clear(); } - /** @param buffer a DirectBuffer backed by off-heap memory Returns address of off-heap memory */ + /** + * @param buffer a DirectBuffer backed by off-heap memory + * @return address of off-heap memory + */ public static long getAddress(ByteBuffer buffer) { return ((DirectBuffer) buffer).address(); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java index a97a2f5ba..b3243d69f 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/RayUtils.java @@ -15,7 +15,7 @@ public class RayUtils { /** * Get all node info from GCS * - *

Returns node info list + * @return node info list */ public static List getAllNodeInfo() { if (Ray.getRuntimeContext().isSingleProcess()) { @@ -28,7 +28,7 @@ public class RayUtils { /** * Get all alive node info map * - *

Returns node info map, key is unique node id , value is node info + * @return node info map, key is unique node id , value is node info */ public static Map getAliveNodeInfoMap() { return getAllNodeInfo().stream() diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ReflectionUtils.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ReflectionUtils.java index bc04a1ded..13a75f8eb 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ReflectionUtils.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ReflectionUtils.java @@ -20,7 +20,7 @@ public class ReflectionUtils { /** * For covariant return type, return the most specific method. * - *

Returns all methods named by {@code methodName}, + * @return all methods named by {@code methodName}, */ public static List findMethods(Class cls, String methodName) { List> classes = new ArrayList<>(); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ResourceUtil.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ResourceUtil.java index b8336cd14..b00b6ee96 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ResourceUtil.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ResourceUtil.java @@ -52,8 +52,8 @@ public class ResourceUtil { } /** - * Returns jvm heap usage ratio. note that one of the survivor space is not include in total - * memory while calculating this ratio. + * @return jvm heap usage ratio. note that one of the survivor space is not include in total + * memory while calculating this ratio. */ public static double getJvmHeapUsageRatio() { Runtime runtime = Runtime.getRuntime(); @@ -61,8 +61,8 @@ public class ResourceUtil { } /** - * Returns jvm heap usage(in bytes). note that this value doesn't include one of the survivor - * space. + * @return jvm heap usage(in bytes). note that this value doesn't include one of the survivor + * space. */ public static long getJvmHeapUsageInBytes() { Runtime runtime = Runtime.getRuntime(); @@ -95,8 +95,8 @@ public class ResourceUtil { } /** - * Returns the system cpu usage. This value is a double in the [0.0,1.0] We will try to use `vsar` - * to get cpu usage by default, and use MXBean if any exception raised. + * @return the system cpu usage. This value is a double in the [0.0,1.0] We will try to use `vsar` + * to get cpu usage by default, and use MXBean if any exception raised. */ public static double getSystemCpuUsage() { double cpuUsage = 0.0; @@ -109,10 +109,10 @@ public class ResourceUtil { } /** - * Returns the "recent cpu usage" for the whole system. This value is a double in the [0.0,1.0] - * interval. A value of 0.0 means that all CPUs were idle during the recent period of time - * observed, while a value of 1.0 means that all CPUs were actively running 100% of the time - * during the recent period being observed + * @return the "recent cpu usage" for the whole system. This value is a double in the [0.0,1.0] + * interval. A value of 0.0 means that all CPUs were idle during the recent period of time + * observed, while a value of 1.0 means that all CPUs were actively running 100% of the time + * during the recent period being observed */ public static double getSystemCpuUtilByMXBean() { return osmxb.getSystemCpuLoad(); @@ -144,7 +144,7 @@ public class ResourceUtil { return cpuUsageFromVsar; } - /** Returnss the system load average for the last minute */ + /** Returns the system load average for the last minute */ public static double getSystemLoadAverage() { return osmxb.getSystemLoadAverage(); } @@ -158,7 +158,8 @@ public class ResourceUtil { * Get containers by hostname of address * * @param containers container list - * @param containerHosts container hostname or address set Returns matched containers + * @param containerHosts container hostname or address set + * @return matched containers */ public static List getContainersByHostname( List containers, Collection containerHosts) { @@ -174,7 +175,8 @@ public class ResourceUtil { /** * Get container by hostname * - * @param hostName container hostname Returns container + * @param hostName container hostname + * @return container */ public static Optional getContainerByHostname( List containers, String hostName) { @@ -188,7 +190,8 @@ public class ResourceUtil { /** * Get container by id * - * @param containerID container id Returns container + * @param containerID container id + * @return container */ public static Optional getContainerById( List containers, ContainerId containerID) { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/JobWorker.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/JobWorker.java index 5a6554802..15200c656 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/JobWorker.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/JobWorker.java @@ -137,8 +137,8 @@ public class JobWorker implements Serializable { /** * Start worker's stream tasks with specific checkpoint ID. * - *

Returns a {@link CallResult} with {@link ChannelRecoverInfo}, contains {@link - * ChannelCreationStatus} of each input queue. + * @return a {@link CallResult} with {@link ChannelRecoverInfo}, contains {@link + * ChannelCreationStatus} of each input queue. */ public CallResult rollback(Long checkpointId, Long startRollbackTs) { synchronized (initialStateChangeLock) { diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java index 5fe774e20..eb48f1691 100644 --- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java +++ b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/util/Mockitools.java @@ -49,8 +49,8 @@ public class Mockitools { /** * Mock get node info map * - * @param nodeInfos all node infos fetched from GCS Returns node info map, key is node unique id, - * value is node info + * @param nodeInfos all node infos fetched from GCS + * @return node info map, key is node unique id, value is node info */ public static Map mockGetNodeInfoMap(List nodeInfos) { return nodeInfos.stream() diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java index 10f99c0b6..921ea8598 100644 --- a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java +++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java @@ -50,8 +50,8 @@ public final class KeyGroupAssignment { * Assigning the key to a key-group index. * * @param key the key to assign. - * @param maxParallelism the maximum parallelism. Returns the key-group index to which the given - * key is assigned. + * @param maxParallelism the maximum parallelism. + * @return the key-group index to which the given key is assigned. */ public static int assignKeyGroupIndexForKey(Object key, int maxParallelism) { return Math.abs(key.hashCode() % maxParallelism); diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java index 933081af5..a632d21d0 100644 --- a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java +++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java @@ -28,7 +28,8 @@ public interface MapState extends UnaryState> { /** * Returns the current value associated with the given key. * - * @param key The key of the mapping Returns The value of the mapping with the given key + * @param key The key of the mapping + * @return The value of the mapping with the given key */ V get(K key); @@ -64,8 +65,8 @@ public interface MapState extends UnaryState> { /** * Returns whether there exists the given mapping. * - * @param key The key of the mapping Returns True if there exists a mapping whose key equals to - * the given key + * @param key The key of the mapping + * @return True if there exists a mapping whose key equals to the given key */ default boolean contains(K key) { return get().containsKey(key); @@ -74,7 +75,7 @@ public interface MapState extends UnaryState> { /** * Returns all the mappings in the state * - *

Returns An iterable view of all the key-value pairs in the state. + * @return An iterable view of all the key-value pairs in the state. */ default Iterable> entries() { return get().entrySet(); @@ -83,7 +84,7 @@ public interface MapState extends UnaryState> { /** * Returns all the keys in the state * - *

Returns An iterable view of all the keys in the state. + * @return An iterable view of all the keys in the state. */ default Iterable keys() { return get().keySet(); @@ -92,7 +93,7 @@ public interface MapState extends UnaryState> { /** * Returns all the values in the state. * - *

Returns An iterable view of all the values in the state. + * @return An iterable view of all the values in the state. */ default Iterable values() { return get().values(); @@ -101,7 +102,7 @@ public interface MapState extends UnaryState> { /** * Iterates over all the mappings in the state. * - *

Returns An iterator over all the mappings in the state + * @return An iterator over all the mappings in the state */ default Iterator> iterator() { return get().entrySet().iterator(); diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java index 5c250b594..637b57314 100644 --- a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java +++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java @@ -24,7 +24,7 @@ public interface UnaryState extends State { /** * get the value in state * - *

Returns the value in state + * @return the value in state */ O get(); }