KAFKA-18434: enrich the authorization error message of connecting to controller (apache#18436)

Reviewers: Divij Vaidya <[email protected]>

Author: FrankYang0529

core/src/main/scala/kafka/server/AuthHelper.scala

@@ -55,7 +55,7 @@ class AuthHelper(authorizer: Option[Authorizer]) {
 
   def authorizeClusterOperation(request: RequestChannel.Request, operation: AclOperation): Unit = {
     if (!authorize(request.context, operation, CLUSTER, CLUSTER_NAME))
-      throw new ClusterAuthorizationException(s"Request $request is not authorized.")
+      throw new ClusterAuthorizationException(s"Request $request needs $operation permission.")
   }
 
   def authorizedOperations(request: RequestChannel.Request, resource: Resource): Int = {

core/src/test/scala/unit/kafka/server/ControllerApisTest.scala

@@ -1286,20 +1286,22 @@ class ControllerApisTest {
 
   @Test
   def testUnauthorizedControllerRegistrationRequest(): Unit = {
-    assertThrows(classOf[ClusterAuthorizationException], () => {
+    val exception = assertThrows(classOf[ClusterAuthorizationException], () => {
       controllerApis = createControllerApis(Some(createDenyAllAuthorizer()), new MockController.Builder().build())
       controllerApis.handleControllerRegistration(buildRequest(
         new ControllerRegistrationRequest(new ControllerRegistrationRequestData(), 0.toShort)))
     })
+    assertTrue(exception.getMessage.contains("needs CLUSTER_ACTION permission"))
   }
 
   @Test
   def testUnauthorizedDescribeClusterRequest(): Unit = {
-    assertThrows(classOf[ClusterAuthorizationException], () => {
+    val exception = assertThrows(classOf[ClusterAuthorizationException], () => {
       controllerApis = createControllerApis(Some(createDenyAllAuthorizer()), new MockController.Builder().build())
       controllerApis.handleDescribeCluster(buildRequest(
         new DescribeClusterRequest(new DescribeClusterRequestData(), 1.toShort)))
     })
+    assertTrue(exception.getMessage.contains("needs ALTER permission"))
   }
 
   @AfterEach