[Cpp worker]Support cpp call java task

cpp/include/ray/api.h

@@ -97,6 +97,9 @@ ray::internal::TaskCaller<F> Task(F func);
 template <typename R>
 ray::internal::TaskCaller<PyFunction<R>> Task(PyFunction<R> func);
 
+template <typename R>
+ray::internal::TaskCaller<JavaFunction<R>> Task(JavaFunction<R> func);
+
 /// Generic version of creating an actor
 /// It is used for creating an actor, such as: ActorCreator<Counter> creator =
 /// ray::Actor(Counter::FactoryCreate<int>).Remote(1);
@@ -232,6 +235,13 @@ inline ray::internal::TaskCaller<PyFunction<R>> Task(PyFunction<R> func) {
  return {ray::internal::GetRayRuntime().get(), std::move(remote_func_holder)};
 }
 
+template <typename R>
+inline ray::internal::TaskCaller<JavaFunction<R>> Task(JavaFunction<R> func) {
+ ray::internal::RemoteFunctionHolder remote_func_holder(
+ "", func.function_name, func.class_name, ray::internal::LangType::JAVA);
+ return {ray::internal::GetRayRuntime().get(), std::move(remote_func_holder)};
+}
+
 inline ray::internal::ActorCreator<JavaActorClass> Actor(JavaActorClass func) {
  ray::internal::RemoteFunctionHolder remote_func_holder(func.module_name,
  func.function_name,

cpp/include/ray/api/task_caller.h

@@ -74,7 +74,7 @@ ObjectRef<boost::callable_traits::return_type_t<F>> TaskCaller<F>::Remote(
  Args &&...args) {
  CheckTaskOptions(task_options_.resources);
 
- if constexpr (is_python_v<F>) {
+ if constexpr (is_x_lang_v<F>) {
  using ArgsTuple = std::tuple<Args...>;
  Arguments::WrapArgs<ArgsTuple>(remote_function_holder_.lang_type,
  &args_,

cpp/include/ray/api/type_traits.h

@@ -72,7 +72,7 @@ template <typename T>
 auto constexpr is_java_v = is_java_t<T>::value;
 
 template <typename T>
-auto constexpr is_x_lang_v = is_java_t<T>::value || is_python_t<T>::value;
+auto constexpr is_x_lang_v = is_java_v<T> || is_python_v<T>;
 
 } // namespace internal
 } // namespace ray

cpp/include/ray/api/xlang_function.h

@@ -59,6 +59,14 @@ struct JavaActorMethod {
  std::string function_name;
 };
 
+template <typename R>
+struct JavaFunction {
+ bool IsJava() { return true; }
+ R operator()() { return {}; }
+ std::string class_name;
+ std::string function_name;
+};
+
 namespace internal {
 
 enum class LangType {

cpp/src/ray/test/cluster/cluster_mode_test.cc

@@ -259,6 +259,12 @@ TEST(RayClusterModeTest, JavaInvocationTest) {
  auto java_actor_ret =
  java_actor_handle.Task(ray::JavaActorMethod<int>{"increase"}).Remote(2);
  EXPECT_EQ(3, *java_actor_ret.Get());
+
+ auto java_task_ret =
+ ray::Task(ray::JavaFunction<std::string>{"io.ray.test.CrossLanguageInvocationTest",
+ "returnInputString"})
+ .Remote("helloworld");
+ EXPECT_EQ("helloworld", *java_task_ret.Get());
 }
 
 TEST(RayClusterModeTest, MaxConcurrentTest) {