Revert "Revert "[Java] Remove RayRuntimeInternal class (#25016)" (#25…

[jovany-wang]

Why are these changes needed?

This reverts commit 804b6b1.

Related issue number

Checks

• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[jovany-wang]

Hi @kira-lin , this PR made the regression of raydp_dataset_test. The error message is

 failure: Lost task 0.3 in stage 7.0 (TID 13) (127.0.0.1 executor 0): java.lang.NoClassDefFoundError: io/ray/runtime/RayRuntimeInternal
	at org.apache.spark.sql.raydp.ObjectStoreWriter.writeToRay(ObjectStoreWriter.scala:66)
	at org.apache.spark.sql.raydp.ObjectStoreWriter.$anonfun$save$2(ObjectStoreWriter.scala:129)
	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
	at org.apache.spark.sql.raydp.ObjectStoreWriter.$anonfun$save$1(ObjectStoreWriter.scala:137)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:863)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:863)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:131)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

We removed the RayRuntimeInternal class which is in the runtime module. But raydp still uses this internal API. https://github.com/oap-project/raydp/blob/c08a7867708224c25a73ae326d9244492b3f71fd/core/src/main/scala/org/apache/spark/sql/raydp/ObjectStoreWriter.scala#L76

We also found the internal API is in used at this line as well https://github.com/oap-project/raydp/blob/c97e68e29c9ac23db8f2aae4ed968df1aefe77c1/core/src/main/java/org/apache/spark/raydp/RayDPUtils.java#L48 , which is not recommended.

@kira-lin Please help to avoid that internal API, thanks.

[kfstorm]

@kira-lin May I know the purpose of holding the owner address?

[kira-lin]

We hold the owner adresses to implement preferredLocations of RDD. @kfstorm

[jovany-wang]

ludwig:test_ludwig is failed on the parent commit https://buildkite.com/ray-project/ray-builders-branch/builds/7758#bbb26405-7d80-4da1-ac02-2b28a8ec898d

[jovany-wang]

windows tests are failed on master https://buildkite.com/ray-project/ray-builders-branch/builds/7802#0180fecf-c0ce-4e43-aad6-66fb6e60485d