Skip to content

[K8s] Rename checkPodAppCanceled to cleanupOrphanPodIfBatchTerminated and improve orphan pod log messages #7620

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

Open
turboFei wants to merge 5 commits into
base: master
Choose a base branch
from
+17 −12
Open

[K8s] Rename checkPodAppCanceled to cleanupOrphanPodIfBatchTerminated and improve orphan pod log messages #7620

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2259cfc
refactor: rename checkPodAppCanceled and improve orphan pod log messages
turboFei Aug 9, 2026
2107bd0
refactor: revert log message changes, keep only method rename
turboFei Aug 9, 2026
7158df3
refactor: improve log message for submit-timeout orphan pod case
turboFei Aug 9, 2026
56425e1
refactor: shorten method name to cleanupOrphanPod
turboFei Aug 9, 2026
46f0df4
refactor: rename cleanupCanceledAppPodExecutor to cleanupOrphanPodExe…
turboFei Aug 9, 2026
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
29 changes: 17 additions & 12 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {

private var expireCleanUpTriggerCacheExecutor: ScheduledExecutorService = _

private var cleanupCanceledAppPodExecutor: ThreadPoolExecutor = _
private var cleanupOrphanPodExecutor: ThreadPoolExecutor = _

private var kubernetesClientInitializeCleanupTerminatedPodExecutor: ThreadPoolExecutor = _

Expand Down Expand Up @@ -208,8 +208,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
cleanupDriverPodCheckInterval,
cleanupDriverPodCheckInterval,
TimeUnit.MILLISECONDS)
cleanupCanceledAppPodExecutor = ThreadUtils.newDaemonCachedThreadPool(
"cleanup-canceled-app-pod-thread")
cleanupOrphanPodExecutor = ThreadUtils.newDaemonCachedThreadPool(
"cleanup-orphan-pod-thread")
kubernetesClientInitializeCleanupTerminatedPodExecutor =
ThreadUtils.newDaemonCachedThreadPool(
"kubernetes-client-initialize-cleanup-terminated-pod-thread")
Expand Down Expand Up @@ -369,9 +369,9 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
expireCleanUpTriggerCacheExecutor = null
}

if (cleanupCanceledAppPodExecutor != null) {
ThreadUtils.shutdown(cleanupCanceledAppPodExecutor)
cleanupCanceledAppPodExecutor = null
if (cleanupOrphanPodExecutor != null) {
ThreadUtils.shutdown(cleanupOrphanPodExecutor)
cleanupOrphanPodExecutor = null
}

if (kubernetesClientInitializeCleanupTerminatedPodExecutor != null) {
Expand All @@ -393,7 +393,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
pod,
appStateSource,
appStateContainer)
checkPodAppCanceled(kubernetesInfo, pod)
cleanupOrphanPod(kubernetesInfo, pod)
}
}

Expand All @@ -414,7 +414,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
appStateSource,
appStateContainer)
if (firstUpdate) {
checkPodAppCanceled(kubernetesInfo, newPod)
cleanupOrphanPod(kubernetesInfo, newPod)
}
}
}
Expand Down Expand Up @@ -577,17 +577,22 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
}
}

private def checkPodAppCanceled(kubernetesInfo: KubernetesInfo, pod: Pod): Unit = {
private def cleanupOrphanPod(kubernetesInfo: KubernetesInfo, pod: Pod): Unit = {
if (kyuubiConf.isRESTEnabled) {
cleanupCanceledAppPodExecutor.submit(new Runnable {
cleanupOrphanPodExecutor.submit(new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
val kyuubiUniqueKey = pod.getMetadata.getLabels.get(LABEL_KYUUBI_UNIQUE_KEY)
val batch = metadataManager.flatMap(_.getBatchSessionMetadata(kyuubiUniqueKey))
if (batch.map(_.state).map(OperationState.withName)
.exists(_ == OperationState.CANCELED)) {
val batchState = batch.map(_.state).map(OperationState.withName)
if (batchState.exists(_ == OperationState.CANCELED)) {
warn(s"[$kubernetesInfo] Batch[$kyuubiUniqueKey] is canceled, " +
s"try to delete the pod ${pod.getMetadata.getName}")
deletePod(kubernetesInfo, pod.getMetadata.getName, kyuubiUniqueKey)
} else if (batchState.exists(_ == OperationState.ERROR) &&
batch.flatMap(_.appState).exists(_ == ApplicationState.NOT_FOUND)) {
warn(s"[$kubernetesInfo] Batch[$kyuubiUniqueKey] failed due to submit timeout" +
s" (app state: NOT_FOUND), try to delete the orphan pod ${pod.getMetadata.getName}")
deletePod(kubernetesInfo, pod.getMetadata.getName, kyuubiUniqueKey)
}
}
})
Expand Down
Loading