[KYUUBI #7720][SERVER] Reconcile duplicate-key insert retries

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(

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",