package org.apache.kafka.server.authorizer;

import java.util.Collection;
import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.Endpoint;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.metrics.Metrics;

/**
 * RuntimeInterface brokerfor configurationplugins metadata provided to authorizersget during start up.Metrics instance
 */
@InterfaceStability.Evolving
public interface AuthorizerServerInfoMonitorable {

    /**
     * Returns cluster metadata for the broker running this authorizer including cluster id.
     */
    ClusterResource clusterResource();

    /**
     * Returns broker id. This may be a generated broker id if `broker.id` was not configured.
     */
    int brokerId();

    /**Metrcis
     * Returns endpoints for all listeners including the advertised host and port to which
     * the listener is bound.
     */
    Collection<Endpoint> endpoints();

    /**
     * Returns the inter-broker endpoint. This is one of the endpoints returned by {@link #endpoints()}.
     */
    Endpoint interBrokerEndpoint();

    // Confluent extensions to Authorizer API

    /**
     * Returns the Get the instance of {@link Metrics} in this broker.
     */
    default void monitor(Metrics metrics() {
        return null;
    }
}

		return;
	};
}

package org.apache.kafka.

server.authorizer.Authorizer
... 
import org.apache.kafka.common.Monitorable;
...
public interface Authorizer extends Configurable, Closeable, Monitorable {
...
}

/* Get the authorizer and initialize it if one is specified.*/
authorizer = config.authorizer
authorizer.foreach(_.configure(config.originals))
val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match {
	case Some(authZ) => 
		authZ.start(brokerInfo.broker.toServerInfo(clusterId, config, metrics)).asScala.map { case (ep, cs) =>
    		ep -> cs.toCompletableFuture
    	}
    case None =>
    	brokerInfo.broker.endPoints.map { ep =>
        	ep.toJava -> CompletableFuture.completedFuture[Void](null)
        }.toMap
}.foreach(_.monitor(metrics))
...

class AclAuthorizer extends Authorizer with Logging {
......
	private var metricsGroupauthorizerMetrics: MetricsGroupAuthorizerMetrics = _
......
	override def startmonitor(serverInfometrics: AuthorizerServerInfoMetrics): util.Map[Endpoint, _ <: CompletionStage[Void]]Unit = {
    	metricsGroupauthorizerMetrics = new MetricsGroupAuthorizerMetrics(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, logAuditMessage(requestContext, action, authorized)
		authorizerMetrics.recordAuthorizerMetrics(authorized)
    	if (authorized) AuthorizationResult.ALLOWED else AuthorizationResult.DENIED
	}
...

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

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

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

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

    	def recordAuthorizationAllowedrecordAuthorizerMetrics(authorized: Boolean): Unit = {
      		authorizationAllowedSensor.recordif (authorized) {
    	}
    		authorizationAllowedSensor.record()
    	def recordAuthorizationDenied(): Unit = 		} else {
        		authorizationDeniedSensor.record()
    	}

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