Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pan3793 wants to merge 5 commits into from
Closed

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

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cbd7afe
Add tests.
JoshRosen Jun 5, 2024
4f85f7e
Implement feature.
JoshRosen Jun 5, 2024
079040c
config doc
JoshRosen Jun 5, 2024
9261a98
version
pan3793 Aug 21, 2025
25cda18
polish log
pan3793 Aug 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Implement feature.
  • Loading branch information
@JoshRosen @pan3793
JoshRosen authored and pan3793 committed Aug 21, 2025
commit 4f85f7e75bb38bf4e1486cdbbe4b1287f557f557
16 changes: 15 additions & 1 deletion core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1022,11 +1022,19 @@ private[spark] class SparkSubmit extends Logging {
e
}

var exitCode: Int = 1
try {
app.start(childArgs.toArray, sparkConf)
exitCode = 0
} catch {
case t: Throwable =>
throw findCause(t)
val cause = findCause(t)
cause match {
case e: SparkUserAppException =>
exitCode = e.exitCode
case _ =>
}
throw cause
} finally {
if (args.master.startsWith("k8s") && !isShell(args.primaryResource) &&
!isSqlShell(args.mainClass) && !isThriftServer(args.mainClass) &&
Expand All @@ -1037,6 +1045,12 @@ private[spark] class SparkSubmit extends Logging {
case e: Throwable => logError("Failed to close SparkContext", e)
}
}
if (sparkConf.get(SUBMIT_CALL_SYSTEM_EXIT_ON_MAIN_EXIT)) {
logInfo(
log"Calling System.exit() with exit code ${MDC(LogKeys.EXIT_CODE, exitCode)} " +
log"because main ${MDC(LogKeys.CONFIG, SUBMIT_CALL_SYSTEM_EXIT_ON_MAIN_EXIT.key)}=true")
Copy link
Contributor

@ForVic ForVic Aug 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

main in this log line feels out of place

Copy link
Member Author

@pan3793 pan3793 Aug 26, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thank you for checking, updated.

exitFn(exitCode)
Copy link

@itskals itskals Aug 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In one of our internal implementations, we initiated a timer that permitted certain slow non-demon threads to complete their tasks, such as updating external metrics and reporting states. Upon the timer's expiration, the system logs a message, records thread dumps of all active threads (this can be used for triaging the cause of the condition), and subsequently exits. Can you extend this patch to do these?

Copy link

@itskals itskals Aug 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Something like this:

`.....
if (forceTerminateJVM) {
createAndStartShutdownThread(timeToWaitInSeconds, exitCode)
}
}

def createAndStartShutdownThread(timeToWaitInSeconds: Long, exitCode: Int): Unit = {
logInfo("Starting the shutdown thread")
val shutdownThread = new Thread(new Runnable {
override def run(): Unit = {
logInfo(s"Shutdown thread will wait for ${timeToWaitInSeconds}s before " +
s"exiting the JVM")
Thread.sleep(TimeUnit.SECONDS.toMillis(timeToWaitInSeconds))
logWarning("There are non-daemon threads preventing this JVM from shutting down")
Thread.getAllStackTraces.keySet().asScala.filter(t => !t.isDaemon && t.isAlive).
foreach(t => {
logWarning(s"== ${t.toString} ==")
logWarning(t.getStackTrace().mkString(""))
})
logWarning(s"Stopping the JVM with System.exit(${exitCode})")
System.exit(exitCode)
}
}, "JVMShutdownThread")
shutdownThread.setDaemon(true)
shutdownThread.start()
}
`

Copy link
Member Author

@pan3793 pan3793 Aug 25, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@itskals I think it's more appropriate to implement this in a shutdown hook. You actually want a graceful shutdown mechanism that's not coupled to either daemon or non-daemon threads. If users want to do some cleanup work or graceful shutdown logic before the JVM terminates, they need to register a shutdown hook rather than creating non-daemon threads.

BTW, you can use three backticks to quote the code block

this is
  a code block

Copy link

@itskals itskals Aug 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While the usage of non-demon threads is not desirable, it's hard to enforce in actual application development.

I think it's more appropriate to implement this in a shutdown hook.
Also possible there.

}
}
}

Expand Down