Skip to content

Commit c6694bd

Browse files
turboFeiclaude
andcommitted
[KYUUBI #7XXX] Conditionally update engineError to clear pending reasons
### Why are the changes needed? In the previous commit [KYUUBI #7313], when a k8s application is pending, we audit the app diagnostics to record the pending reason. This information is eventually updated to the `engineError` field in the metadata store via the `updateMetadata` method. However, when the application transitions from pending to finished successfully, if there is no error (i.e., `engineError` is `None`), the old pending reason remains in the metadata, causing confusion. ### What are the changes? Modified `JDBCMetadataStore.updateMetadata()` to conditionally update the `engineError` field: - Previously: Only updated when `engineError` was `Some(error)` - Now: Updates when either: 1. `engineError` is defined (Some), OR 2. `engineId` is defined (indicating app has been launched) This ensures that: - Pending reasons are properly cleared when the application transitions to success (engineId is set) - engineError won't be updated unnecessarily when neither engineError nor engineId is present in the update ### How was this patch tested? - Enhanced unit test `update engineError conditionally based on engineError or engineId presence` in `JDBCMetadataStoreSuite` - The test verifies three scenarios: 1. Setting engineError when it's defined works correctly 2. Clearing engineError when engineId is defined (even if engineError is None) 3. engineError remains unchanged when neither engineError nor engineId is in the update 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 2037881 commit c6694bd

File tree

2 files changed

+66
-2
lines changed

2 files changed

+66
-2
lines changed

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -376,9 +376,11 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
376376
setClauses += "engine_state = ?"
377377
params += metadata.engineState
378378
}
379-
metadata.engineError.foreach { error =>
379+
// Update engineError when it's defined or when engineId is defined
380+
// This ensures pending reasons are cleared when app transitions to success
381+
if (metadata.engineError.isDefined || Option(metadata.engineId).isDefined) {
380382
setClauses += "engine_error = ?"
381-
params += error
383+
params += metadata.engineError.orNull
382384
}
383385
if (metadata.peerInstanceClosed) {
384386
setClauses += "peer_instance_closed = ?"

kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStoreSuite.scala

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -271,6 +271,68 @@ class JDBCMetadataStoreSuite extends KyuubiFunSuite {
271271
assert(jdbcMetadataStore.getLatestSchemaUrl(Seq(url1, url2, url3, url4, url5)).get === url5)
272272
}
273273

274+
test("update engineError conditionally based on engineError or engineId presence") {
275+
val batchId = UUID.randomUUID().toString
276+
val kyuubiInstance = "localhost:10099"
277+
val batchMetadata = Metadata(
278+
identifier = batchId,
279+
sessionType = SessionType.BATCH,
280+
realUser = "kyuubi",
281+
username = "kyuubi",
282+
ipAddress = "127.0.0.1",
283+
kyuubiInstance = kyuubiInstance,
284+
state = "PENDING",
285+
resource = "intern",
286+
className = "org.apache.kyuubi.SparkWC",
287+
requestName = "kyuubi_batch",
288+
requestConf = Map("spark.master" -> "local"),
289+
requestArgs = Seq("100"),
290+
createTime = System.currentTimeMillis(),
291+
engineType = "spark",
292+
clusterManager = Some("local"))
293+
294+
jdbcMetadataStore.insertMetadata(batchMetadata)
295+
296+
// Case 1: Update engineError when engineError is defined
297+
val pendingMetadata = batchMetadata.copy(
298+
state = "PENDING",
299+
engineError = Some("Pod pending: Insufficient CPU"))
300+
jdbcMetadataStore.updateMetadata(pendingMetadata)
301+
302+
var retrievedMetadata = jdbcMetadataStore.getMetadata(batchId)
303+
assert(retrievedMetadata.engineError == Some("Pod pending: Insufficient CPU"))
304+
305+
// Case 2: When app transitions to running with engineId, engineError should be cleared
306+
val runningMetadata = pendingMetadata.copy(
307+
state = "RUNNING",
308+
engineId = "app-123",
309+
engineError = None)
310+
jdbcMetadataStore.updateMetadata(runningMetadata)
311+
312+
retrievedMetadata = jdbcMetadataStore.getMetadata(batchId)
313+
assert(retrievedMetadata.engineError == None)
314+
assert(retrievedMetadata.engineId == "app-123")
315+
316+
// Case 3: Update without engineError and without engineId should not update engineError
317+
// First set an error again
318+
val errorMetadata = runningMetadata.copy(engineError = Some("New error"))
319+
jdbcMetadataStore.updateMetadata(errorMetadata)
320+
retrievedMetadata = jdbcMetadataStore.getMetadata(batchId)
321+
assert(retrievedMetadata.engineError == Some("New error"))
322+
323+
// Now update state without engineError and without engineId - error should remain
324+
val stateOnlyUpdate = Metadata(
325+
identifier = batchId,
326+
state = "FINISHED")
327+
jdbcMetadataStore.updateMetadata(stateOnlyUpdate)
328+
329+
retrievedMetadata = jdbcMetadataStore.getMetadata(batchId)
330+
assert(retrievedMetadata.engineError == Some("New error")) // Should remain unchanged
331+
332+
// Clean up
333+
jdbcMetadataStore.cleanupMetadataByIdentifier(batchId)
334+
}
335+
274336
test("kubernetes engine info") {
275337
val tag = UUID.randomUUID().toString
276338
val metadata = KubernetesEngineInfo(

0 commit comments

Comments
 (0)