[SPARK-48547][DEPLOY] Add opt-in flag to have SparkSubmit automatically call System.exit after user code main method exits

[JoshRosen]

What changes were proposed in this pull request?

This PR adds a new SparkConf flag option, spark.submit.callSystemExitOnMainExit (default false), which when true will cause SparkSubmit to call System.exit() in the JVM once the user code's main method has exited (for Java / Scala jobs) or once the user's Python or R script has exited.

Why are the changes needed?

This is intended to address a longstanding issue where spark-submit runs might hang after user code has completed:

According to Java’s java.lang.Runtime docs:

The Java Virtual Machine initiates the shutdown sequence in response to one of several events:

1. when the number of live non-daemon threads drops to zero for the first time (see note below on the JNI Invocation API);
2. when the Runtime.exit or System.exit method is called for the first time; or
3. when some external event occurs, such as an interrupt or a signal is received from the operating system.

For Python and R programs, SparkSubmit’s PythonRunner and RRunner will call System.exit() if the user program exits with a non-zero exit code (see python and R runner code).

But for Java and Scala programs, plus any successful R or Python programs, Spark will not automatically call System.exit.

In those situation, the JVM will only shutdown when, via event (1), all non-daemon threads have exited (unless the job is cancelled and sent an external interrupt / kill signal, corresponding to event (3)).

Thus, non-daemon threads might cause logically-completed spark-submit jobs to hang rather than completing.

The non-daemon threads are not always under Spark's own control and may not necessarily be cleaned up by SparkContext.stop().

Thus, it is useful to have an opt-in functionality to have SparkSubmit automatically call System.exit() upon main method exit (which usually, but not always, corresponds to job completion): this option will allow users and data platform operators to enforce System.exit() calls without having to modify individual jobs' code.

Does this PR introduce any user-facing change?

Yes, it adds a new user-facing configuration option for opting in to a behavior change.

How was this patch tested?

New tests in SparkSubmitSuite, including one which hangs (failing with a timeout) unless the new option is set to true.

Was this patch authored or co-authored using generative AI tooling?

No.

[pan3793]

I think this patch also covers the SPARK-34674 and SPARK-42698 cases. And maybe the code introduced in SPARK-34674 can be removed.

SPARK-34674 is used to tackle the Spark on K8s exit issues, it also introduces side-effects:

• the shutdown process is different between K8s and other platforms - for Spark jobs the user does not call spark.stop explicitly, K8s shutdowns SparkContext ahead, and relies on JVM shutdown hooks to handle other shutdown logic; other platforms just rely on JVM shutdown hooks
• as SPARK-42219 pointed out, SparkContext always shutdown after the Main method has finished.

In SPARK-42698, a similar shutdown logic was requested for the YARN client mode to propagate the exit code to AM.

[JoshRosen]

I think this patch also covers the SPARK-34674 and SPARK-42698 cases. And maybe the code introduced in SPARK-34674 can be removed.

Thanks for the pointers. I was aware of SPARK-34674 for Spark on Kubernetes (I spotted its logic when editing SparkSubmit) but was previously unaware of the SPARK-42698 AM request.

This raises an interesting question of whether my new flag should also be calling sc.stop() if it was not stopped by a user. If I don't invoke it (and Spark isn't running on YARN) then the SparkContext's own shutdown hook at

spark/core/src/main/scala/org/apache/spark/SparkContext.scala

Lines 691 to 704 in 7cba1ab

	// Make sure the context is stopped if the user forgets about it. This avoids leaving
	// unfinished event logs around after the JVM exits cleanly. It doesn't help if the JVM
	// is killed, though.
	logDebug("Adding shutdown hook") // force eager creation of logger
	_shutdownHookRef = ShutdownHookManager.addShutdownHook(
	ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY) { () =>
	logInfo("Invoking stop() from shutdown hook")
	try {
	stop()
	} catch {
	case e: Throwable =>
	logWarning("Ignoring Exception while stopping SparkContext from shutdown hook", e)
	}
	}

would stop the context, but that gives us less explicit control around the relative ordering of the SparkContext stop and any other cleanup activities performed by user- or third-party library shutdown hooks (some of which might be implicitly (and possibly incorrectly) assuming that the SparkContext will have already been stopped before their hooks run). Explicitly stopping the SparkContext would have the advantage of letting us propagate the exit code (as was proposed in #40314).

If I do that, though, then my new proposed configuration's name might need to change to reflect the more general behavior (right now spark.submit.callSystemExitOnMainExit is very literally and narrowly named for exactly what it's doing).

There's also a question of defaults and user-facing behavior changes: although "stop the context then call System.exit()" is probably a reasonable behavior for most cases, there might be a long-tail of less common use-cases where users explicitly don't want any automatic cleanup or shutdown.

Today, for better or worse, we have a set of on-by-default cleanup behaviors for Spark on Kubernetes and I'm slightly hesitant to want to change those defaults for fear of breaking someone's use-case. For example, maybe a user wants the SparkContext to be automatically stopped on job completion but doesn't want System.exit() to be called (e.g. because they're using the org.apache.spark.launcher package's InProcessLauncher and maybe they have their own cleanup / post action logic that must run after the child application, and in such cases a coupling of the auto-stop option with System.exit could cause problems for them. On the other hand, though, in such a niche use-case they could have their own "framework" code stop the SparkContext themselves (i.e. disable all of the automatic cleanup logic then handle their own niche use case in their own custom logic).

Maybe we should optimize for some sensible "batteries included" top-level cleanup option which can be completely disabled by users who want to handle the cleanup themselves, but, when enabled, does a reasonably linear standard cleanup flow?

[pan3793]

This raises an interesting question of whether my new flag should also be calling sc.stop() if it was not stopped by a user.

although "stop the context then call System.exit()" is probably a reasonable behavior for most cases, there might be a long-tail of less common use-cases where users explicitly don't want any automatic cleanup or shutdown.

I believe System.exit and SparkContext.stop are different things, and we'd better have two flags to cover all possible cases, hopefully. Of course, if both flags are enabled, call SparkContext.stop first then System.exit.

... shutdown hooks (some of which might be implicitly (and possibly incorrectly) assuming that the SparkContext will have already been stopped before their hooks run).

... and versus, I do see that some users write shutdown hooks rely on the priority to assume that the execution happens before/after SparkContext shutdown.

I'm slightly hesitant to want to change those defaults for fear of breaking someone's use-case.

Yeah, I understand. Well, on the other hand, Spark 4.0 may be a good opportunity(at least better than minor versions like 4.1, 4.2 ...) to make such a breaking change with a more reasonable and consistent(as mentioned above, the default K8s's shutdown behavior is different from other platforms), as long as it has flags to allow users change back to previous behaviors.

non-daemon threads might cause logically-completed spark-submit jobs to hang rather than completing.

And IIRC, in Spark on YARN cluster mode(the most popular cases in my company and customers), the AM will call System.exit thus there are no such issues, but it does happen on K8s mode, this indeed introduces extra efforts for Spark workloads migration from YARN to K8s.

[pan3793]

... using the org.apache.spark.launcher package's InProcessLauncher

As InProcessLauncher was mentioned, let me share some use cases.

We do see perf benefits of InProcessLauncher in Kyuubi(a gateway for Spark and other engines) use cases. Well, seems it is not well maintained today, though I suppose it should be a public API and an official approach to launch Spark jobs. For example, due to SPARK-41006, it does not work for K8s mode.

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

[pan3793]

Hi @JoshRosen, do you have a plan to move this PR forward? I consistently get reports from users who migrate Spark workloads from YARN to K8s complaining that daemon threads created by user codes block the driver JVM exit ...