KAFKA-19315: Move ControllerMutationQuotaManager to server module

checkstyle/import-control-server.xml

@@ -86,6 +86,7 @@
     <allow pkg="org.apache.kafka.server" />
     <allow pkg="org.apache.kafka.image" />
     <allow pkg="org.apache.kafka.network.metrics" />
+    <allow pkg="org.apache.kafka.network" />
     <allow pkg="org.apache.kafka.storage.internals.log" />
     <allow pkg="org.apache.kafka.storage.internals.checkpoint" />
     <allow pkg="org.apache.logging.log4j" />

core/src/main/java/kafka/server/ClientRequestQuotaManager.java

@@ -25,17 +25,16 @@
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Rate;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.ClientQuotaManager;
 import org.apache.kafka.server.config.ClientQuotaManagerConfig;
 import org.apache.kafka.server.quota.ClientQuotaCallback;
 import org.apache.kafka.server.quota.QuotaType;
 import org.apache.kafka.server.quota.QuotaUtils;
 
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
-import scala.jdk.javaapi.CollectionConverters;
-import scala.jdk.javaapi.OptionConverters;
-
 @SuppressWarnings("this-escape")
 public class ClientRequestQuotaManager extends ClientQuotaManager {
     // Since exemptSensor is for all clients and has a constant name, we do not expire exemptSensor and only
@@ -56,7 +55,7 @@ public ClientRequestQuotaManager(
             String threadNamePrefix, 
             Optional<Plugin<ClientQuotaCallback>> quotaCallbackPlugin
     ) {
-        super(config, metrics, QuotaType.REQUEST, time, threadNamePrefix, OptionConverters.toScala(quotaCallbackPlugin));
+        super(config, metrics, QuotaType.REQUEST, time, threadNamePrefix, quotaCallbackPlugin);
         this.maxThrottleTimeMs = TimeUnit.SECONDS.toMillis(config.quotaWindowSizeSeconds);
         this.metrics = metrics;
         this.exemptMetricName = metrics.metricName("exempt-request-time", QuotaType.REQUEST.toString(), "Tracking exempt-request-time utilization percentage");
@@ -72,8 +71,8 @@ private void recordExempt(double value) {
     }
 
     /**
-     * Records that a user/clientId changed request processing time being throttled. If quota has been violated, return
-     * throttle time in milliseconds. Throttle time calculation may be overridden by sub-classes.
+     * Records that a user/clientId changed request processing time being throttled. If the quota has been violated, return
+     * throttle time in milliseconds. Subclasses may override throttle time calculation.
      * @param request client request
      * @return Number of milliseconds to throttle in case of quota violation. Zero otherwise
      */
@@ -103,8 +102,8 @@ public long throttleTime(QuotaViolationException e, long timeMs) {
     }
 
     @Override
-    public MetricName clientQuotaMetricName(scala.collection.immutable.Map<String, String> quotaMetricTags) {
-        return metrics.metricName("request-time", QuotaType.REQUEST.toString(), "Tracking request-time per user/client-id", CollectionConverters.asJava(quotaMetricTags));
+    public MetricName clientQuotaMetricName(Map<String, String> quotaMetricTags) {
+        return metrics.metricName("request-time", QuotaType.REQUEST.toString(), "Tracking request-time per user/client-id", quotaMetricTags);
     }
 
     private double nanosToPercentage(long nanos) {

core/src/main/java/kafka/server/QuotaFactory.java

@@ -21,6 +21,8 @@
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.ClientQuotaManager;
+import org.apache.kafka.server.ControllerMutationQuotaManager;
 import org.apache.kafka.server.config.ClientQuotaManagerConfig;
 import org.apache.kafka.server.config.QuotaConfig;
 import org.apache.kafka.server.config.ReplicationQuotaManagerConfig;
@@ -29,8 +31,6 @@
 
 import java.util.Optional;
 
-import scala.Option;
-import scala.jdk.javaapi.OptionConverters;
 
 public class QuotaFactory {
 
@@ -124,13 +124,12 @@ public static QuotaManagers instantiate(
         String role
     ) {
         Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin = createClientQuotaCallback(cfg, metrics, role);
-        Option<Plugin<ClientQuotaCallback>> clientQuotaCallbackPluginOption = OptionConverters.toScala(clientQuotaCallbackPlugin);
 
         return new QuotaManagers(
-            new ClientQuotaManager(clientConfig(cfg), metrics, QuotaType.FETCH, time, threadNamePrefix, clientQuotaCallbackPluginOption),
-            new ClientQuotaManager(clientConfig(cfg), metrics, QuotaType.PRODUCE, time, threadNamePrefix, clientQuotaCallbackPluginOption),
+            new ClientQuotaManager(clientConfig(cfg), metrics, QuotaType.FETCH, time, threadNamePrefix, clientQuotaCallbackPlugin),
+            new ClientQuotaManager(clientConfig(cfg), metrics, QuotaType.PRODUCE, time, threadNamePrefix, clientQuotaCallbackPlugin),
             new ClientRequestQuotaManager(clientConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallbackPlugin),
-            new ControllerMutationQuotaManager(clientControllerMutationConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallbackPluginOption),
+            new ControllerMutationQuotaManager(clientControllerMutationConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallbackPlugin),
             new ReplicationQuotaManager(replicationConfig(cfg), metrics, QuotaType.LEADER_REPLICATION, time),
             new ReplicationQuotaManager(replicationConfig(cfg), metrics, QuotaType.FOLLOWER_REPLICATION, time),
             new ReplicationQuotaManager(alterLogDirsReplicationConfig(cfg), metrics, QuotaType.ALTER_LOG_DIRS_REPLICATION, time),

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

@@ -34,6 +34,7 @@ import org.apache.kafka.coordinator.group.GroupCoordinator
 import org.apache.kafka.coordinator.share.ShareCoordinator
 import org.apache.kafka.coordinator.transaction.TransactionLogConfig
 import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, NodeToControllerChannelManager}
+import org.apache.kafka.server.ControllerMutationQuota;
 
 import scala.collection.{Map, Seq, Set, mutable}
 import scala.jdk.CollectionConverters._