KAFKA-19100: Use ProcessRole instead of String in AclApis (#19406)

Use the ProcessRole enum instead of hardcoding the role

Reviewers: Mickael Maison <[email protected]>, PoAn Yang
<[email protected]>, Jhen-Yung Hsu <[email protected]>, Ken Huang
<[email protected]>

Author: gongxuanzhang

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

@@ -30,6 +30,7 @@ import org.apache.kafka.common.requests._
 import org.apache.kafka.common.resource.Resource.CLUSTER_NAME
 import org.apache.kafka.common.resource.ResourceType
 import org.apache.kafka.security.authorizer.AuthorizerUtils
+import org.apache.kafka.server.ProcessRole
 import org.apache.kafka.server.authorizer._
 
 import java.util
@@ -45,9 +46,9 @@ import scala.jdk.OptionConverters.RichOptional
 class AclApis(authHelper: AuthHelper,
               authorizer: Option[Authorizer],
               requestHelper: RequestHandlerHelper,
-              name: String,
+              role: ProcessRole,
               config: KafkaConfig) extends Logging {
-  this.logIdent = "[AclApis-%s-%s] ".format(name, config.nodeId)
+  this.logIdent = "[AclApis-%s-%s] ".format(role, config.nodeId)
   private val alterAclsPurgatory =
       new DelayedFuturePurgatory(purgatoryName = "AlterAcls", brokerId = config.nodeId)
 

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

@@ -56,6 +56,7 @@ import org.apache.kafka.image.publisher.ControllerRegistrationsPublisher
 import org.apache.kafka.metadata.{BrokerHeartbeatReply, BrokerRegistrationReply}
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.server.ProcessRole
 import org.apache.kafka.server.authorizer.Authorizer
 import org.apache.kafka.server.common.{ApiMessageAndVersion, RequestLocal}
 
@@ -84,7 +85,7 @@ class ControllerApis(
   val configHelper = new ConfigHelper(metadataCache, config, metadataCache)
   val requestHelper = new RequestHandlerHelper(requestChannel, quotas, time)
   val runtimeLoggerManager = new RuntimeLoggerManager(config.nodeId, logger.underlying)
-  private val aclApis = new AclApis(authHelper, authorizer, requestHelper, "controller", config)
+  private val aclApis = new AclApis(authHelper, authorizer, requestHelper, ProcessRole.ControllerRole, config)
 
   def isClosed: Boolean = aclApis.isClosed
 

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

@@ -58,7 +58,7 @@ import org.apache.kafka.common.{Node, TopicIdPartition, TopicPartition, Uuid}
 import org.apache.kafka.coordinator.group.{Group, GroupConfigManager, GroupCoordinator}
 import org.apache.kafka.coordinator.share.ShareCoordinator
 import org.apache.kafka.metadata.{ConfigRepository, MetadataCache}
-import org.apache.kafka.server.ClientMetricsManager
+import org.apache.kafka.server.{ClientMetricsManager, ProcessRole}
 import org.apache.kafka.server.authorizer._
 import org.apache.kafka.server.common.{GroupVersion, RequestLocal, TransactionVersion}
 import org.apache.kafka.server.share.context.ShareFetchContext
@@ -113,7 +113,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   val configHelper = new ConfigHelper(metadataCache, config, configRepository)
   val authHelper = new AuthHelper(authorizer)
   val requestHelper = new RequestHandlerHelper(requestChannel, quotas, time)
-  val aclApis = new AclApis(authHelper, authorizer, requestHelper, "broker", config)
+  val aclApis = new AclApis(authHelper, authorizer, requestHelper, ProcessRole.BrokerRole, config)
   val configManager = new ConfigAdminManager(brokerId, config, configRepository)
   val describeTopicPartitionsRequestHandler = new DescribeTopicPartitionsRequestHandler(
     metadataCache, authHelper, config)