[Java] Fix some potential bugs about Ray.shutdown() (#5693)

java/runtime/src/main/java/org/ray/runtime/RayDevRuntime.java

@@ -33,6 +33,10 @@ public class RayDevRuntime extends AbstractRayRuntime {
 
   @Override
   public void shutdown() {
+    if (taskSubmitter != null) {
+      ((LocalModeTaskSubmitter) taskSubmitter).shutdown();
+      taskSubmitter = null;
+    }
     taskExecutor = null;
   }
 

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

@@ -133,6 +133,7 @@ public final class RayNativeRuntime extends AbstractRayRuntime {
     }
     if (null != manager) {
       manager.cleanup();
+      manager = null;
     }
   }
 

java/runtime/src/main/java/org/ray/runtime/task/LocalModeTaskSubmitter.java

@@ -210,6 +210,10 @@ public class LocalModeTaskSubmitter implements TaskSubmitter {
     }
   }
 
+  public void shutdown() {
+    exec.shutdown();
+  }
+
   public static ActorId getActorId(TaskSpec taskSpec) {
     ByteString actorId = null;
     if (taskSpec.getType() == TaskType.ACTOR_CREATION_TASK) {

java/test/src/main/java/org/ray/api/test/BaseMultiLanguageTest.java

@@ -45,13 +45,17 @@ public abstract class BaseMultiLanguageTest {
     }
   }
 
-  @BeforeClass(alwaysRun = true)
-  public void setUp() {
+  private void checkMultiLanguageTestFlag() {
     if (!"1".equals(System.getenv("ENABLE_MULTI_LANGUAGE_TESTS"))) {
       LOGGER.info("Skip Multi-language tests because environment variable "
           + "ENABLE_MULTI_LANGUAGE_TESTS isn't set");
       throw new SkipException("Skip test.");
     }
+  }
+
+  @BeforeClass(alwaysRun = true)
+  public void setUp() {
+    checkMultiLanguageTestFlag();
 
     // Delete existing socket files.
     for (String socket : ImmutableList.of(RAYLET_SOCKET_NAME, PLASMA_STORE_SOCKET_NAME)) {
@@ -102,6 +106,8 @@ public abstract class BaseMultiLanguageTest {
 
   @AfterClass(alwaysRun = true)
   public void tearDown() {
+    checkMultiLanguageTestFlag();
+
     // Disconnect to the cluster.
     Ray.shutdown();
     System.clearProperty("ray.redis.address");

java/test/src/main/java/org/ray/api/test/BaseTest.java

@@ -14,7 +14,7 @@ public class BaseTest {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(BaseTest.class);
 
-  private List<File> filesToDelete;
+  private List<File> filesToDelete = ImmutableList.of();
 
   @BeforeMethod(alwaysRun = true)
   public void setUpBase(Method method) {