diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala index 567dccfbc94..3d79889d26f 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala @@ -135,6 +135,18 @@ class MetadataManager extends AbstractService("MetadataManager") { } } + // Verifies whether a duplicate-key error on a queued insert retry actually means the + // insert already succeeded earlier (e.g. the acknowledgement was lost), by checking + // whether a matching row already exists. Used only by the retry loop below, so the + // fail-fast contract of insertMetadata() itself is unchanged for direct/foreground calls. + private def verifyInsertPostcondition(metadata: Metadata): Boolean = { + Option(_metadataStore.getMetadata(metadata.identifier)).exists { existing => + existing.identifier == metadata.identifier && + existing.sessionType == metadata.sessionType && + existing.createTime == metadata.createTime + } + } + def getBatch(batchId: String): Option[Batch] = { getBatchSessionMetadata(batchId).map(buildBatch) } @@ -344,11 +356,30 @@ class MetadataManager extends AbstractService("MetadataManager") { info(s"Retrying metadata requests for $id") var request = ref.metadataRequests.peek() while (request != null) { - request match { - case insert: InsertMetadata => - insertMetadata(insert.metadata, asyncRetryOnError = false) - case update: UpdateMetadata => - updateMetadata(update.metadata, asyncRetryOnError = false) + try { + request match { + case insert: InsertMetadata => + insertMetadata(insert.metadata, asyncRetryOnError = false) + case update: UpdateMetadata => + updateMetadata(update.metadata, asyncRetryOnError = false) + } + } catch { + // A duplicate-key error while retrying a queued insert means a row + // with this identifier already exists, e.g. the earlier attempt + // actually succeeded but its acknowledgement was lost. Verify the + // postcondition instead of leaving this request stuck at the head + // of the queue forever, which would also block all later requests + // for this session (KYUUBI #7720). + case rethrow: Throwable + if request.isInstanceOf[InsertMetadata] && + unrecoverableDBErr(rethrow) && + verifyInsertPostcondition(request.metadata) => + warn( + s"Insert for ${request.metadata.identifier} returned a " + + "duplicate-key error on retry but a matching row already " + + "exists; treating as idempotent success and removing it " + + "from the retry queue.", + rethrow) } ref.metadataRequests.remove(request) MetricsSystem.tracing(_.markMeter( diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/MetadataManagerSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/MetadataManagerSuite.scala index de2e651f44a..71f9a7233af 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/MetadataManagerSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/MetadataManagerSuite.scala @@ -97,6 +97,30 @@ class MetadataManagerSuite extends KyuubiFunSuite { } } + test("[KYUUBI #7720] duplicate-key insert retry does not block the retry queue") { + withMetadataManager(Map( + METADATA_REQUEST_ASYNC_RETRY_ENABLED.key -> "true", + METADATA_REQUEST_RETRY_INTERVAL.key -> "100")) { metadataManager => + val metadata = newMetadata() + // Insert the row directly, simulating an earlier attempt that actually succeeded. + metadataManager.insertMetadata(metadata) + + // Queue a retry of the *same* insert, simulating a replay whose earlier + // acknowledgement was lost. Before the fix, this would throw a duplicate-key + // error on every retry attempt and never be removed from the queue. + metadataManager.addMetadataRetryRequest(InsertMetadata(metadata)) + // Queue an unrelated update behind it, to confirm it is no longer blocked. + val metadataToUpdate = metadata.copy(state = "RUNNING") + metadataManager.addMetadataRetryRequest(UpdateMetadata(metadataToUpdate)) + + val retryRef = metadataManager.getMetadataRequestsRetryRef(metadata.identifier) + eventually(timeout(3.seconds)) { + assert(!retryRef.hasRemainingRequests()) + assert(metadataManager.getBatch(metadata.identifier).map(_.getState).contains("RUNNING")) + } + } + } + test("async metadata request metrics") { withMetadataManager(Map( METADATA_REQUEST_ASYNC_RETRY_ENABLED.key -> "true",