class AclAuthorizer extends Authorizer with Logging {
......
	private var metricsGroup: MetricsGroup = _
......
	override def start(serverInfo: AuthorizerServerInfo): util.Map[Endpoint, _ <: CompletionStage[Void]] = {
    	metricsGroup = new MetricsGroup(serverInfo.metrics())
    	serverInfo.endpoints.asScala.map { endpoint =>
      	endpoint -> CompletableFuture.completedFuture[Void](null) }.toMap.asJava
	}
......

	private def authorizeAction(requestContext: AuthorizableRequestContext, action: Action): AuthorizationResult = {
		......
		// Record metrics
    	if (authorized) {
      		metricsGroup.recordAuthorizationAllowed()
    	} else {
      		metricsGroup.recordAuthorizationDenied()
    	}
   	 	metricsGroup.recordAuthorizationRequest()
		logAuditMessage(requestContext, action, authorized)
    	if (authorized) AuthorizationResult.ALLOWED else AuthorizationResult.DENIED
	}

	class MetricsGroupAuthorizerMetrics(metrics: Metrics) {
    	val GROUP_NAME = "acl-authorizer-kafka.security.authorizer.metrics"
    	val authorizationAllowedSensor = metrics.sensor("aclauthorizer-authorization-allowed")
    	authorizationAllowedSensor.add(metrics.metricName("authorization-allowed-rate-per-hourminute", GROUP_NAME,
      		"The number of authoization allowed per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount()))

    	val authorizationDeniedSensor = metrics.sensor("aclauthorizer-authorization-denied")
    	authorizationDeniedSensor.add(metrics.metricName("authorization-denied-rate-per-hourminute", GROUP_NAME,
      		"The number of authoization denied per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount()))

    	val authorizationRequestSensor = metrics.sensor("aclauthorizer-authorization-request")
    	authorizationRequestSensor.add(metrics.metricName("authorization-request-rate-per-hourminute", GROUP_NAME,
      		"The number of authoization request per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount()))

    	metrics.addMetric(metrics.metricName("aclacls-total-count", GROUP_NAME, "The number of acls defined"),
      	(config, now) => aclCache.size)

    	def recordAuthorizationAllowed(): Unit = {
      		authorizationAllowedSensor.record()
    	}

    	def recordAuthorizationDenied(): Unit = {
      		authorizationDeniedSensor.record()
    	}

    	def recordAuthorizationRequest(): Unit = {
      		authorizationRequestSensor.record()
    	}
  	}
}