check admin

Author: turboFei

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -568,14 +568,20 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
     content = Array(new Content(
       mediaType = MediaType.APPLICATION_JSON,
       schema = new Schema(implementation = classOf[ReassignBatchResponse]))),
-    description = "Reassign batches to a new instance")
+    description =
+      "Reassign batch sessions on an unreachable kyuubi instance to the current kyuubi instance")
   @POST
   @Path("/reassign")
   @Consumes(Array(MediaType.APPLICATION_JSON))
   def reassignBatchSessions(request: ReassignBatchRequest): ReassignBatchResponse = {
+    val userName = fe.getSessionUser(Map.empty[String, String])
+    val ipAddress = fe.getIpAddress
     val kyuubiInstance = request.getKyuubiInstance
     val newKyuubiInstance = fe.connectionUrl
-    info(s"Received request to reassign batch sessions from $kyuubiInstance to $newKyuubiInstance")
+    info(s"Received reassign $kyuubiInstance batch sessions request from $userName/$ipAddress")
+    if (!fe.isAdministrator(userName)) {
+      throw new ForbiddenException(s"$userName is not allowed to reassign the batches")
+    }
     if (kyuubiInstance == newKyuubiInstance) {
       throw new IllegalStateException(s"KyuubiInstance is alive: $kyuubiInstance")
     }