dongjoon-hyun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4129910495
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
+ preKubernetesResources, driverPodName, preResourcesApplied.get(),
podCreatedByUs.get()))
var watch: Watch = null
var createdDriverPod: Pod = null
try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
- throw e
- }
+ try {
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ preResourcesApplied.set(true)
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create [resource]\"
first." +
+ " It should be yes. And please also check your feature step
implementation.")
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ throw e
+ }
- // Refresh all pre-resources' owner references
- try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.pods().resource(createdDriverPod).delete()
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
+ try {
+ createdDriverPod =
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ podCreatedByUs.set(true)
+ } catch {
+ case NonFatal(e) =>
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ throw e
+ }
+
+ // Refresh all pre-resources' owner references
+ try {
+ addOwnerReference(createdDriverPod, preKubernetesResources)
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ } catch {
+ case NonFatal(e) =>
+ kubernetesClient.pods().resource(createdDriverPod).delete()
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ throw e
+ }
+ } finally {
+ // Past this point, the pre-resources either have an owner reference
(success) or have
+ // already been explicitly deleted by one of the catch blocks above
(failure) -- either
+ // way, the shutdown hook has nothing left to do, so remove it instead
of leaving a no-op
+ // hook registered for the remaining lifetime of this process.
+ shutdownHookOps.removeHook(cleanupHookRef)
Review Comment:
This `finally` removes the cleanup hook even when the pre-resources were not
cleaned up.
- A catch block's own `delete()` throws. For example, the connection drops
during `create()`, and L234 fails for the same reason. This is the case
mentioned in the previous review: "a dropped connection where the catch
cleanup's delete also fails".
- If the pod deletion at L245 throws, L246 is skipped.
- A fatal error (e.g. `OutOfMemoryError`, `StackOverflowError`,
`LinkageError`) doesn't match `case NonFatal(e)`, so no catch block runs, but
this `finally` still removes the hook.
In these cases, the conf ConfigMap and the
keytab/delegation-token/credentials Secrets remain ownerless forever. So, the
comments at L201-202 ("or a fatal JVM error") and L250-253 ("already been
explicitly deleted by one of the catch blocks above") are not accurate either.
Since the deletion ignores 404, it would be safer to remove the hook only on
the success path (after L242), or only after a catch block's cleanup succeeds.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
+ preKubernetesResources, driverPodName, preResourcesApplied.get(),
podCreatedByUs.get()))
var watch: Watch = null
var createdDriverPod: Pod = null
try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
- throw e
- }
+ try {
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ preResourcesApplied.set(true)
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create [resource]\"
first." +
+ " It should be yes. And please also check your feature step
implementation.")
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
Review Comment:
`KubernetesClientUtils.configMapNameDriver` is a per-JVM `val`, so all
submissions from the same JVM (e.g. via `InProcessLauncher`) share this
ConfigMap name. Now that it is a pre-resource, this catch block deletes it by
name when the apply fails. If app B is submitted while app A's driver pod is
still pending, B's apply fails with 422 (the ConfigMap is immutable and its
data differs), and this line deletes A's live ConfigMap. As a result, A's
driver is stuck with `configmap ... not found`. On master, B also fails, but
the post-resource catch block deletes only B's pod.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala:
##########
@@ -74,7 +74,9 @@ private[spark] class PodTemplateConfigMapStep(conf:
KubernetesConf)
}
}
- override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+ // SPARK-38079: this config map is mounted by the driver pod, so it must be
created as a
+ // pre-resource (before the pod itself) to avoid a "configmap ... not found"
mount race.
+ override def getAdditionalPreKubernetesResources(): Seq[HasMetadata] = {
Review Comment:
This also changes the output of `KubernetesDriverBuilder` (`@DeveloperApi`),
which `apache/spark-kubernetes-operator` consumes. The operator's `AppInitStep`
creates the pre-resources before the driver pod without any owner reference,
and `AppCleanUpStep` deletes the pre-resources only for a `SchedulingFailure`
attempt. So, if the operator stops abruptly between the pre-resource creation
and the pod creation, and the `SparkApplication` is deleted in the meantime,
the pod template ConfigMap (common via `executorSpec.podTemplateSpec`) and the
credentials/Kerberos Secrets are never deleted. Previously, they were
post-resources created with the driver pod as the owner.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -209,6 +293,54 @@ private[spark] class Client(
log"submission ID ${MDC(SUBMISSION_ID, sId)} into Kubernetes")
}
}
+
+ // SPARK-38079: best-effort cleanup for pre-resources (and the driver pod,
if it was created)
+ // that were left without an owner reference because this process was
terminated abruptly
+ // before the owner-reference refresh in run() completed. See the comment
above the shutdown
+ // hook registration in run() for why this window exists and why it must
stop mattering once
+ // that refresh completes.
+ //
+ // package-private (rather than private) so it can be unit-tested directly
-- actually
+ // triggering a JVM shutdown hook from a test is impractical, so tests
instead call this
+ // method directly with injected pre/post-state and a fake
recoveryClientFactory.
+ private[submit] def cleanupOrphanedPreResources(
+ preResources: Seq[HasMetadata],
+ driverPodName: String,
+ preResourcesApplied: Boolean,
+ podCreatedByUs: Boolean): Unit = {
+ if (preResourcesApplied) {
+ try {
+ Utils.tryWithResource(recoveryClientFactory()) { recoveryClient =>
Review Comment:
This hook runs blocking network I/O without any deadline inside Spark's
single Hadoop-managed shutdown hook. It builds a new client (kubeconfig
auto-configuration, possibly with an exec credential plugin) and deletes the
resources sequentially. If the API server becomes unreachable after the first
apply (e.g. a VPN drop) and the user presses Ctrl-C while `create()` hangs,
each DELETE can take ~40s (10s connection timeout with fabric8 retries). So,
`spark-submit` hangs until Hadoop's shutdown hook timeout (30s by default)
cancels it. A second Ctrl-C is ignored during the shutdown, and lower-priority
hooks such as the temp directory deletion are delayed. Previously, Ctrl-C
exited immediately. Could you bound this with a short request timeout without
retries, or with an explicit deadline?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
Review Comment:
In terms of the trade-off, this moves the
credentials/keytab/delegation-token Secrets and the conf ConfigMap (plain-text
`spark.properties`) into a window without an owner reference, which doesn't
exist on master. SIGKILL, OOM kill, node loss, or the grace period expiry
between L218 and L242 (where no shutdown hook runs) leaves them behind forever,
without any labels to find them. On the other hand, the benefit is to remove a
transient `FailedMount`. The kubelet retries missing ConfigMap/Secret volumes
with a backoff (up to about 2 minutes) and doesn't fail the pod, so the claim
that this race "can cause the driver pod to fail to start altogether" doesn't
seem to be accurate. The live test in the PR description also reports only
`FailedMount` events.
Have you considered `spec.schedulingGates` (GA since K8s 1.30, while Spark
requires K8s >= 1.35)? Creating the driver pod first with a scheduling gate,
applying all resources once with the owner reference, and then removing the
gate would avoid the ownerless window, the shutdown hook, the second apply, and
the `OwnerReferencesPermissionEnforcement` issue altogether. A pod template
with a fixed `spec.nodeName` would need a fallback.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -209,6 +293,54 @@ private[spark] class Client(
log"submission ID ${MDC(SUBMISSION_ID, sId)} into Kubernetes")
}
}
+
+ // SPARK-38079: best-effort cleanup for pre-resources (and the driver pod,
if it was created)
+ // that were left without an owner reference because this process was
terminated abruptly
+ // before the owner-reference refresh in run() completed. See the comment
above the shutdown
+ // hook registration in run() for why this window exists and why it must
stop mattering once
+ // that refresh completes.
+ //
+ // package-private (rather than private) so it can be unit-tested directly
-- actually
+ // triggering a JVM shutdown hook from a test is impractical, so tests
instead call this
+ // method directly with injected pre/post-state and a fake
recoveryClientFactory.
+ private[submit] def cleanupOrphanedPreResources(
+ preResources: Seq[HasMetadata],
+ driverPodName: String,
+ preResourcesApplied: Boolean,
+ podCreatedByUs: Boolean): Unit = {
+ if (preResourcesApplied) {
+ try {
+ Utils.tryWithResource(recoveryClientFactory()) { recoveryClient =>
+ try {
+ recoveryClient.resourceList(preResources: _*).delete()
Review Comment:
fabric8's `resourceList(...).delete()` deletes the items sequentially and
stops at the first non-404 exception. Since the driver conf ConfigMap is
appended last (L194), any failure in front of it skips it. For example, with
`VolcanoFeatureStep`, if the submitter cannot delete `podgroups` (403), the
conf ConfigMap (plain-text `spark.properties`) remains ownerless with only one
warning log. The same applies to the catch blocks at L224, L234, and L246.
Could you delete each item independently?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
+ preKubernetesResources, driverPodName, preResourcesApplied.get(),
podCreatedByUs.get()))
var watch: Watch = null
var createdDriverPod: Pod = null
try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
- throw e
- }
+ try {
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ preResourcesApplied.set(true)
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create [resource]\"
first." +
+ " It should be yes. And please also check your feature step
implementation.")
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ throw e
+ }
- // Refresh all pre-resources' owner references
- try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.pods().resource(createdDriverPod).delete()
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
+ try {
+ createdDriverPod =
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ podCreatedByUs.set(true)
+ } catch {
+ case NonFatal(e) =>
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
Review Comment:
Since the driver conf ConfigMap is always a pre-resource now, this
`delete()` is no longer a no-op for default submissions (on master,
`driverPreKubernetesResources` is empty unless Volcano or a user feature step
is used). If it throws, its exception replaces the original `e`, and the
`logError` hint in the next line is not printed.
For example, if the pod creation is rejected by `ResourceQuota`, PodSecurity
admission, or a webhook, and the submitter has no `delete` permission on
ConfigMaps (the docs only require pod permissions), users will see only `cannot
delete resource "configmaps"` instead of the real cause. The same applies to
L224 and L246. Could you guard these cleanup deletions (e.g. `case NonFatal(t)
=> e.addSuppressed(t)`) and log the hint before the cleanup?
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -227,6 +230,203 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf2key=conf2value"))
}
+ test("SPARK-38079: driver's own config map is created before the driver pod,
" +
+ "to avoid a mount race") {
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+ submissionClient.run()
+ // The first resourceList(...) call is always the pre-resources
application, which
+ // happens before the driver pod is created. The driver's own config map
must be
+ // included there so that it exists in Kubernetes before the pod that
mounts it.
+ val firstResourceListCall =
createdResourcesArgumentCaptor.getAllValues.get(0)
+ val configMaps = firstResourceListCall.filter(_.isInstanceOf[ConfigMap])
+ assert(configMaps.nonEmpty,
+ "the driver's own config map must be sent as a pre-resource, before the
driver pod " +
+ "is created, to avoid a \"configmap ... not found\" mount race
(SPARK-38079)")
+
+ // Safety check: the pre-resource owner-reference refresh (the second
resourceList()
+ // call) must still set an owner reference on the config map, same as
before this
+ // change, so that it is still garbage-collected along with the driver pod.
+ val secondResourceListCall =
createdResourcesArgumentCaptor.getAllValues.get(1)
+ val refreshedConfigMap = secondResourceListCall
+ .filter(_.isInstanceOf[ConfigMap]).map(_.asInstanceOf[ConfigMap]).head
+ val ownerReferences = refreshedConfigMap.getMetadata.getOwnerReferences
+ assert(ownerReferences.size() === 1)
+ assert(ownerReferences.get(0).getName === POD_NAME)
+ assert(ownerReferences.get(0).getUid === DRIVER_POD_UID)
+ }
+
+ // SPARK-38079: making the driver's own config map (and other
credential-bearing resources,
+ // e.g. Kerberos keytab/delegation-token secrets) a pre-resource above means
they briefly
+ // exist without an owner reference (see the comment on
cleanupOrphanedPreResources in
+ // KubernetesClientApplication.scala). run() registers a shutdown hook to
best-effort clean
+ // those up if this process is terminated abruptly in that window. Actually
triggering a JVM
+ // shutdown hook from a test is impractical, so these tests instead call the
(package-private)
+ // cleanup method directly, the same way the hook itself would.
+ private def newRecoveryClientMocks(): (KubernetesClient, RESOURCE_LIST,
PodResource) = {
+ val recoveryClient = mock[KubernetesClient]
+ val recoveryResourceList = mock[RESOURCE_LIST]
+ val recoveryPodOperations = mock[PODS]
+ val recoveryPodsWithNamespace = mock[PODS_WITH_NAMESPACE]
+ val recoveryNamedPod = mock[PodResource]
+ doReturn(recoveryResourceList)
+
.when(recoveryClient).resourceList(ArgumentMatchers.any[Array[HasMetadata]]():
_*)
+ when(recoveryClient.pods()).thenReturn(recoveryPodOperations)
+
when(recoveryPodOperations.inNamespace(kconf.namespace)).thenReturn(recoveryPodsWithNamespace)
+
when(recoveryPodsWithNamespace.withName(POD_NAME)).thenReturn(recoveryNamedPod)
+ (recoveryClient, recoveryResourceList, recoveryNamedPod)
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup deletes orphaned pre-resources and
the driver pod " +
+ "if this submission created it") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
true)
+
+ verify(recoveryResourceList).delete()
+ verify(recoveryNamedPod).delete()
+ // The recovery client is built fresh for this cleanup and must not leak.
+ verify(recoveryClient).close()
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup does not delete the driver pod if
this submission " +
+ "never created it") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ // podCreatedByUs = false: e.g. the pod creation API call itself never
succeeded (or a
+ // differently-submitted application happens to be using the same pod
name), so this
+ // cleanup must not delete a pod it did not create.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
false)
+
+ verify(recoveryResourceList).delete()
+ verify(recoveryNamedPod, never()).delete()
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup is a no-op if pre-resources were
never applied") {
+ var recoveryClientFactoryInvoked = false
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => {
+ recoveryClientFactoryInvoked = true
+ mock[KubernetesClient]
+ }))
+
+ // preResourcesApplied = false: the pre-resource serverSideApply() call
itself never
+ // succeeded, so there is nothing to have been left orphaned -- and in
particular, the
+ // existing catch block for that call (in run()) has already handled
cleanup of whatever
+ // partial state that failed call may have left behind.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = false, podCreatedByUs =
false)
+
+ assert(!recoveryClientFactoryInvoked,
+ "the recovery client must not be built at all when there is nothing to
clean up")
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup swallows exceptions from the
recovery client " +
+ "(best-effort only)") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ doThrow(new RuntimeException("simulated API server failure"))
+ .when(recoveryResourceList).delete()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ // Must not throw: this runs on the JVM shutdown-hook thread, where an
uncaught exception
+ // would only be printed to stderr and otherwise has no one left to
meaningfully handle it.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
true)
+
+ // The pod delete is independent of the (failed) pre-resource delete
above, and must still
+ // be attempted.
+ verify(recoveryNamedPod).delete()
+ }
+
+ // SPARK-38079: the tests above all call cleanupOrphanedPreResources
directly, since actually
+ // triggering a JVM shutdown hook from a test is impractical. That leaves
the wiring in run()
+ // itself -- does it register a hook before applying pre-resources, and
remove that exact hook
+ // once done with them -- uncovered by those tests alone. These two tests
close that gap by
+ // injecting a fake ShutdownHookOps instead.
+ test("SPARK-38079: cleanup hook is registered before pre-resources are
applied and " +
+ "removed once run() completes successfully") {
+ var registeredHook: Option[() => Unit] = None
+ var removedRef: Option[Any] = None
+ val fakeOps = ShutdownHookOps(
+ addHook = { hook =>
+ registeredHook = Some(hook)
+ "fake-hook-ref"
+ },
+ removeHook = { ref =>
+ removedRef = Some(ref)
+ true
+ })
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ shutdownHookOpsOverride = Some(fakeOps))
+
+ submissionClient.run()
+
+ assert(registeredHook.isDefined,
Review Comment:
These new tests don't verify the core behaviors.
- The hook lambda registered by `run()` is never invoked. L394 checks only
`registeredHook.isDefined`, and the failure test discards the hook. So,
swapping the two `.get()` arguments at L212, replacing the hook body with `()
=> ()`, or dropping the conf ConfigMap from the list would still pass.
- No test verifies the main client's `resourceList(...).delete()` when the
pod creation fails (L234).
- The test at L244 checks only that the first `resourceList` call contains a
ConfigMap, not that it happens before `namedPods.create()` (e.g. with
`InOrder`). Also, `addOwnerReference` mutates the captured instance in place,
so the owner reference check at L253-259 cannot tell the state at the time of
each call.
- `ArgumentMatchers.any[Array[HasMetadata]]()` at L276 matches exactly one
vararg element in Mockito 5. With two or more pre-resources, the stub returns
`null`, and the NPE is swallowed by `NonFatal`.
`ArgumentMatchers.any(classOf[Array[HasMetadata]])` matches the whole array.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/HadoopConfDriverFeatureStep.scala:
##########
@@ -112,7 +112,9 @@ private[spark] class HadoopConfDriverFeatureStep(conf:
KubernetesConf)
}
}
- override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+ // SPARK-38079: this config map is mounted by the driver pod, so it must be
created as a
+ // pre-resource (before the pod itself) to avoid a "configmap ... not found"
mount race.
+ override def getAdditionalPreKubernetesResources(): Seq[HasMetadata] = {
Review Comment:
Since this step emits the ConfigMap whenever `HADOOP_CONF_DIR` is set in the
environment (`System.getenv`), the two `SPARK-37331` tests in
`KubernetesDriverBuilderSuite` now depend on the environment. If
`HADOOP_CONF_DIR` is set (even to a non-existent path), the size assertions at
L72 (`=== 0`) and L85 (`=== 1`) fail, because the forked test JVM inherits the
variable. Previously, this ConfigMap went to `driverKubernetesResources`, which
those tests don't check. Excluding `HadoopConfDriverFeatureStep` via
`KUBERNETES_DRIVER_POD_EXCLUDED_FEATURE_STEPS` in those tests would fix it.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala:
##########
@@ -225,7 +225,10 @@ private[spark] class
KerberosConfDriverFeatureStep(kubernetesConf: KubernetesDri
}
}
- override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+ // SPARK-38079: the krb5 config map, keytab secret, and delegation token
secret below are
+ // all mounted by the driver pod, so they must be created as pre-resources
(before the pod
+ // itself) to avoid a "configmap/secret ... not found" mount race.
+ override def getAdditionalPreKubernetesResources(): Seq[HasMetadata] = {
Review Comment:
Another corner case for `apache/spark-kubernetes-operator`: when
`AppInitStep` runs again after the driver pod was created (e.g. the status
update to `DriverRequested` failed, or the operator restarted), it rebuilds the
spec and force-applies all pre-resources (`AppInitStep.java:137`). Since the
delegation token Secret is immutable and a new `KerberosConfDriverFeatureStep`
obtains new tokens, the apply fails with 422 (`field is immutable`). This leads
to `SchedulingFailure`, which deletes the running driver. Previously, this
Secret was a driver resource, which is only get-or-created. This requires a
Kerberos-enabled operator JVM, so it is rare.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
+ preKubernetesResources, driverPodName, preResourcesApplied.get(),
podCreatedByUs.get()))
var watch: Watch = null
var createdDriverPod: Pod = null
try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
- throw e
- }
+ try {
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ preResourcesApplied.set(true)
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create [resource]\"
first." +
+ " It should be yes. And please also check your feature step
implementation.")
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ throw e
+ }
- // Refresh all pre-resources' owner references
- try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.pods().resource(createdDriverPod).delete()
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
+ try {
+ createdDriverPod =
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ podCreatedByUs.set(true)
+ } catch {
+ case NonFatal(e) =>
+ kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ throw e
+ }
+
+ // Refresh all pre-resources' owner references
+ try {
+ addOwnerReference(createdDriverPod, preKubernetesResources)
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
Review Comment:
This refresh now sets `ownerReferences` on objects that already exist (the
driver conf ConfigMap on every submission, plus the moved ConfigMaps/Secrets),
so the API server admits it as an UPDATE. Under the
`OwnerReferencesPermissionEnforcement` admission plugin, which OpenShift
enables by default, an UPDATE that changes `ownerReferences` requires `delete`
permission on the object (`cannot set an ownerRef on a resource you can't
delete` in `plugin/pkg/admission/gc/gc_admission.go`). On master, these
resources are created once with the owner reference already set, which is a
CREATE, and that check is skipped.
As a result, a submitter that can create/patch but cannot delete ConfigMaps
now fails on every submission. For example, the default controller RBAC of
kubeflow/spark-operator v2.5.2, which runs `spark-submit` in the controller
pod, has `get/list/watch/create/update/patch` on `configmaps` but no `delete`.
1. L218 creates the ConfigMap without an owner reference (CREATE, allowed).
2. L242 fails with 403.
3. The catch block deletes the pod, and then L246 also fails with 403 on the
ConfigMap deletion, which replaces the original exception.
4. The `finally` removes the hook, so an ownerless ConfigMap is left behind
on every attempt.
This contradicts "No behavior change for successful submissions" in the PR
description. Note that `docs/running-on-kubernetes.md` only requires pod
permissions for the submitter.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
+ preKubernetesResources, driverPodName, preResourcesApplied.get(),
podCreatedByUs.get()))
var watch: Watch = null
var createdDriverPod: Pod = null
try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
- throw e
- }
+ try {
+ kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+ preResourcesApplied.set(true)
Review Comment:
The flags are set only after the blocking calls return (L219, L231), and the
hook reads them only once when it starts (L211-212), while the main thread
keeps running during the JVM shutdown.
- If SIGINT/SIGTERM arrives while this `serverSideApply()` is in flight
(fabric8 applies the list items one by one), the hook sees
`preResourcesApplied=false` and does nothing. However, the items already
applied (e.g. the credentials Secret, which comes first) remain ownerless
forever.
- If the signal arrives while `create()` (L230) is in flight, the hook reads
`(true, false)` and deletes only the pre-resources. When `create()` succeeds
afterward, the driver pod is stuck in `ContainerCreating`. This contradicts the
comment at L204 ("and the driver pod, if this submission created it").
Since the deletion ignores 404, you can set each flag before the call, and
let the hook read the flags after building its client (or coordinate with the
main thread).
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
Review Comment:
The comments are too verbose compared to the surrounding code. The main code
adds 75 comment lines vs 71 code lines, and the same rationale (the ownerless
window, "triggering a JVM shutdown hook from a test is impractical") is
repeated 5-6 times. Some of them are inaccurate or will become stale after
merging.
- L205 "It is a no-op ... once the owner-reference refresh has completed":
both flags are `true` after the refresh, and the hook becomes harmless only
because the `finally` removes it.
- L208 "today's behavior", `ClientSuite` L214 "is now a pre-resource", and
`ClientSuite` L423 "unchanged by SPARK-38079" describe the diff rather than the
code.
- `ClientSuite` L357-358 "would only be printed to stderr" is not true,
because `SparkShutdownHookManager.runAll()` logs it via
`Utils.logUncaughtExceptions`.
- The same warning message is duplicated at L324 and L340.
For reference, the executor-side fix for the same race (SPARK-38794) used a
single 3-line comment. Could you keep one concise comment here and trim the
others?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -136,43 +188,75 @@ private[spark] class Client(
val driverPodName = resolvedDriverPod.getMetadata.getName
// setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
- try {
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
- } catch {
- case NonFatal(e) =>
- logError("Please check \"kubectl auth can-i create [resource]\"
first." +
- " It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod itself, to avoid a
"configmap ... not
+ // found" mount race between the driver pod and the config map it depends
on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
They are created
+ // here without an owner reference -- the owner reference is only added
once the driver pod
+ // exists (see "Refresh all pre-resources' owner references" below), since
Kubernetes owner
+ // references require the owner's UID, which does not exist before the pod
is created. If
+ // this process is terminated abruptly in that window (e.g. Ctrl-C,
SIGTERM, or a fatal JVM
+ // error), the pre-resources would otherwise be silently orphaned in the
namespace forever,
+ // since Kubernetes only garbage-collects via owner references. This
shutdown hook makes a
+ // best-effort attempt to delete them (and the driver pod, if this
submission created it) in
+ // that case. It is a no-op -- and removed entirely, see the `finally`
below -- once the
+ // owner-reference refresh has completed; from that point on the existing
owner references
+ // make normal Kubernetes garbage collection sufficient, and e.g. Ctrl-C
while waiting for
+ // the application to complete must keep today's behavior of just
detaching.
+ val preResourcesApplied = new AtomicBoolean(false)
+ val podCreatedByUs = new AtomicBoolean(false)
+ val cleanupHookRef = shutdownHookOps.addHook(() =>
cleanupOrphanedPreResources(
Review Comment:
The shutdown hook commit (+326/-30) is about four times larger than the fix
itself (+75/-24), but it covers only a part of the window. It covers none of
SIGKILL, OOM, and "a dropped connection where the catch cleanup's delete also
fails" (see the other comments). In addition, the same ownerless window has
existed for `VolcanoFeatureStep`'s `PodGroup` since SPARK-37331 (Apache Spark
3.3.0) without such a hook. I'd suggest taking the other option in the previous
review ("accept the window and state the trade-off in the PR description"), or
handling this in a separate JIRA issue.
BTW, the PR description needs to be updated. `VolcanoFeatureStep` already
uses `getAdditionalPreKubernetesResources()`, so "no feature step was using it"
is not true. The shutdown hook is not mentioned at all, and the test count
(386) is outdated.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -94,12 +113,45 @@ private[spark] object ClientArguments {
* implemented features.
* @param kubernetesClient the client to talk to the Kubernetes API server
* @param watcher a watcher that monitors and logs the application status
+ * @param recoveryClientFactoryOverride overrides how the SPARK-38079
shutdown-hook cleanup path
+ * (see run()) builds the fresh
Kubernetes client it uses to
+ * best-effort delete pre-resources left
orphaned by an
+ * abrupt termination. Not reused from
`kubernetesClient`
+ * because that one may already be
closed, or concurrently
+ * in use, by the time the hook runs.
Exposed only for
+ * testing; `None` (the default) builds a
real client the
+ * same way
KubernetesClientApplication.run() builds
+ * `kubernetesClient`. A plain default
value can't itself
+ * call `conf.sparkConf` here, since
`conf` is a
+ * constructor parameter, not yet a class
member, at the
+ * point default values are evaluated.
+ * @param shutdownHookOpsOverride overrides how the SPARK-38079 cleanup hook
(see run()) is
+ * registered/removed. Exposed only for
testing -- actually
+ * triggering a JVM shutdown hook from a test
is impractical, so
+ * tests instead inject fakes here to assert
that run() registers
+ * a hook before applying pre-resources and
removes that exact
+ * hook once it is done with them (success or
failure). `None`
+ * (the default) delegates to the real
`ShutdownHookManager`.
*/
private[spark] class Client(
conf: KubernetesDriverConf,
builder: KubernetesDriverBuilder,
kubernetesClient: KubernetesClient,
- watcher: LoggingPodStatusWatcher) extends Logging {
+ watcher: LoggingPodStatusWatcher,
+ recoveryClientFactoryOverride: Option[() => KubernetesClient] = None,
+ shutdownHookOpsOverride: Option[ShutdownHookOps] = None) extends Logging {
+
+ private val recoveryClientFactory: () => KubernetesClient =
recoveryClientFactoryOverride
Review Comment:
This duplicates the client creation in `KubernetesClientApplication.run()`
(L375-384), so both can diverge later (e.g. the auth prefix or the client
type), and the difference would show up only in the untested abrupt-termination
path. The reason for not reusing `kubernetesClient` (L119-121, "may already be
closed, or concurrently in use") doesn't hold except for a narrow race, because
the hook is removed before `run()` returns, i.e. within the
`Utils.tryWithResource` lifetime of `kubernetesClient`, and a fabric8 client
can be shared across threads. Building a new client during the shutdown also
re-runs the kubeconfig auto-configuration (and an exec credential plugin like
`aws eks get-token`), which adds to the shutdown delay. If we keep the hook,
reusing `kubernetesClient` (or `kubernetesClient.newClient(RequestConfig)` with
a short timeout) would remove this override and its 12-line Scaladoc.
Similarly, `ShutdownHookOps` doesn't need to be an `Option`. A plain default
parameter w
ould be enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
