JVM exit hangs when the application returns from main without SparkContext.stop()

[zhangfengcdt]

Describe the bug

#4734 (fix for #4725) makes the global tokio runtime releasable and calls release_runtime() from CometDriverPlugin.shutdown() / CometExecutorPlugin.shutdown(). That fixes the hang whenever SparkContext.stop() runs. It does not fix the case where an application simply returns from main (or a PySpark driver exits) without calling spark.stop() — which is a very common shape for batch jobs. In that case the JVM still hangs forever in DestroyJavaVM, exactly as described in #4725.

The reason is an ordering problem that #4734 cannot get around on its own:

1. Tokio runtime threads that call back into the JVM (memory pool acquireMemory, ScanExec fed from a JVM iterator, CometScalarSubquery, …) attach lazily and permanently via jni-rs, using AttachCurrentThread. That makes them non-daemon JVM threads.
2. When main returns, the launcher calls DestroyJavaVM. HotSpot's Threads::destroy_vm() waits until it is the last non-daemon thread before it runs shutdown hooks (invoke_shutdown_hooks() comes after the wait).
3. Spark's ShutdownHookManager is what would call SparkContext.stop() → plugin shutdown() → release_runtime(). So the runtime can only be released by a hook that can only run after the runtime's threads have exited. Circular; the JVM waits forever.

Without an explicit SparkContext.stop() the release in #4734 is unreachable.

Steps to reproduce

The precondition is simply "at least one tokio runtime thread has called into the JVM", which any workload with JVM callbacks from spawned work satisfies. Sketch (upstream-only triggers):

// spark-submit / plain `java` main; Comet plugin + native exec enabled;
// spark.memory.offHeap.enabled=true (unified pool → JVM acquireMemory callbacks)
val spark = SparkSession.builder().master("local[2]")
  .config("spark.plugins", "org.apache.spark.CometPlugin")
  .config("spark.sql.extensions", "org.apache.comet.CometSparkSessionExtensions")
  .config("spark.comet.enabled", "true")
  .config("spark.comet.exec.enabled", "true")
  .config("spark.memory.offHeap.enabled", "true")
  .config("spark.memory.offHeap.size", "256m")
  .getOrCreate()
spark.range(1000000).write.mode("overwrite").parquet(path)
val df = spark.read.parquet(path)
  .selectExpr("sum(id)")
  .where("1 <= (SELECT max(id) FROM parquet.`" + path + "`)") // scalar subquery → JVM call from native
df.collect()
require(df.queryExecution.executedPlan.toString.contains("Comet"))
println("main returning")
// intentionally NO spark.stop()

Run it as a real process and wait: the process prints main returning and never exits. jstack shows DestroyJavaVM waiting plus N unnamed non-daemon threads ("Thread-30", "Thread-31", … — one per runtime worker) with no Java frames; /proc//task/*/comm identifies them as tokio-rt-worker.

Expected behavior

A Spark application that returns from main without SparkContext.stop() exits, the same way it does with Comet disabled. Spark's own thread pools are all daemon precisely so that this works; Comet's runtime threads are currently the exception.

Additional context

• Comet main after feat: release tokio runtime on driver/executor exit #4734 (1.0.0); jni 0.22.4; observed on Linux (JDK 17, Spark 4.1) and macOS (JDK 11, Spark 3.5).
• The upstream development.md threading section already documents that attachment is cached until the worker exits and that "the tokio runtime has to be shut down for the JVM to be able to exit" — this issue is the case where that shutdown can never be initiated.