Skip to content

[SPARK-51660][CORE] Gracefully handle when MDC is not supported #50452

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
robreeves wants to merge 4 commits into from
Closed

[SPARK-51660][CORE] Gracefully handle when MDC is not supported #50452

Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e5d3e00
update mdc check
robreeves Mar 28, 2025
f5dcf72
style fix
robreeves Mar 28, 2025
775e982
new approach that doesnt have side effect
robreeves Mar 29, 2025
894ac8a
moved to companion object
robreeves Mar 31, 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
23 changes: 17 additions & 6 deletions core/src/main/scala/org/apache/spark/executor/Executor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,21 @@ private[spark] class Executor(
// finish downloading dependencies.
private val updateDependenciesLock = new ReentrantLock()

private lazy val mdcIsSupported = {
try {
// This tests if any class initialization error is thrown
val testKey = System.nanoTime().toString
MDC.put(testKey, "testValue")
MDC.remove(testKey)

true
} catch {
case t: Throwable =>
logInfo("MDC is not supported.", t)
false
}
}

// No ip or host:port - just hostname
Utils.checkHost(executorHostname)
// must not have port specified.
Expand Down Expand Up @@ -930,21 +945,17 @@ private[spark] class Executor(
}

private def setMDCForTask(taskName: String, mdc: Seq[(String, String)]): Unit = {
try {
if (mdcIsSupported) {
mdc.foreach { case (key, value) => MDC.put(key, value) }
// avoid overriding the takName by the user
MDC.put(taskNameMDCKey, taskName)
} catch {
case _: NoSuchFieldError => logInfo("MDC is not supported.")
}
}

private def cleanMDCForTask(taskName: String, mdc: Seq[(String, String)]): Unit = {
try {
if (mdcIsSupported) {
mdc.foreach { case (key, _) => MDC.remove(key) }
MDC.remove(taskNameMDCKey)
} catch {
case _: NoSuchFieldError => logInfo("MDC is not supported.")
}
}

Expand Down