KAFKA-18399 Remove ZooKeeper from KafkaApis (6/N): USER_SCRAM_CREDENTIALS

Author: frankvicky

Diff for: core/src/main/scala/kafka/server/KafkaApis.scala

@@ -3370,10 +3370,7 @@ class KafkaApis(val requestChannel: RequestChannel,
     } else {
       metadataSupport match {
         case ZkSupport(adminManager, controller, zkClient, forwardingManager, metadataCache, _) =>
-          val result = adminManager.describeUserScramCredentials(
-            Option(describeUserScramCredentialsRequest.data.users).map(_.asScala.map(_.name).toList))
-          requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-            new DescribeUserScramCredentialsResponse(result.setThrottleTimeMs(requestThrottleMs)))
+         throw KafkaApis.shouldNeverReceive(request)
         case RaftSupport(_, metadataCache) =>
           val result = metadataCache.describeScramCredentials(describeUserScramCredentialsRequest.data())
           requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
@@ -3383,21 +3380,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   }
 
   def handleAlterUserScramCredentialsRequest(request: RequestChannel.Request): Unit = {
-    val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request))
-    val alterUserScramCredentialsRequest = request.body[AlterUserScramCredentialsRequest]
-
-    if (!zkSupport.controller.isActive) {
-      requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-        alterUserScramCredentialsRequest.getErrorResponse(requestThrottleMs, Errors.NOT_CONTROLLER.exception))
-    } else if (authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
-      val result = zkSupport.adminManager.alterUserScramCredentials(
-        alterUserScramCredentialsRequest.data.upsertions().asScala, alterUserScramCredentialsRequest.data.deletions().asScala)
-      requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-        new AlterUserScramCredentialsResponse(result.setThrottleTimeMs(requestThrottleMs)))
-    } else {
-      requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-        alterUserScramCredentialsRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
-    }
+    throw KafkaApis.shouldAlwaysForward(request)
   }
 
   def handleAlterPartitionRequest(request: RequestChannel.Request): Unit = {