KAFKA-20650: Share ClientQuotaCallback across broker and controller in combined mode

clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java

@@ -97,6 +97,12 @@ default T deserialize(String topic, Headers headers, byte[] data) {
      * <p>Similarly, if this method is overridden, the implementation cannot make any assumptions about the
      * passed in {@link ByteBuffer} either.
      *
+     * <p>The serialized bytes are the buffer's {@link ByteBuffer#remaining() remaining} bytes, i.e. those
+     * between its current {@code position} and {@code limit}. An implementation should read the data without
+     * changing the buffer's {@code position}, {@code limit}, or {@code mark}: the caller retains ownership of
+     * the buffer and may read it again after this method returns (for example, to determine the serialized
+     * size of the value), so mutating the buffer's state can corrupt unrelated processing.
+     *
      * <p>It is recommended to deserialize a {@code null} {@link ByteBuffer} to a {@code null} object.
      *
      * <p>Note that the passed in {@link Headers} may be empty, but never {@code null}.

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

@@ -103,7 +103,31 @@ public static QuotaManagers instantiate(
         );
     }
 
-    private static Optional<Plugin<ClientQuotaCallback>> createClientQuotaCallback(
+    /**
+     * Create {@link QuotaManagers} from a pre-created {@link ClientQuotaCallback} plugin instead of
+     * instantiating a new one. This lets a combined-mode broker and controller share a single callback
+     * instance (see KAFKA-20650) rather than each creating their own via reflection with divergent state.
+     */
+    public static QuotaManagers instantiate(
+        KafkaConfig cfg,
+        Metrics metrics,
+        Time time,
+        String threadNamePrefix,
+        Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin
+    ) {
+        return new QuotaManagers(
+            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, 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),
+            clientQuotaCallbackPlugin
+        );
+    }
+
+    public static Optional<Plugin<ClientQuotaCallback>> createClientQuotaCallback(
         KafkaConfig cfg, 
         Metrics metrics, 
         String role

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

@@ -208,7 +208,7 @@ class BrokerServer(
       val clientTelemetryExporterPlugin = new ClientTelemetryExporterPlugin()
 
       config.dynamicConfig.initialize(Some(clientTelemetryExporterPlugin))
-      quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-", ProcessRole.BrokerRole.toString)
+      quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-", sharedServer.clientQuotaCallbackPlugin)
       DynamicBrokerConfig.readDynamicBrokerConfigsFromSnapshot(raftManager, config, quotaManagers, logContext)
 
       /* start scheduler */

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

@@ -283,7 +283,7 @@ class ControllerServer(
       quotaManagers = QuotaFactory.instantiate(config,
         metrics,
         time,
-        s"controller-${config.nodeId}-", ProcessRole.ControllerRole.toString)
+        s"controller-${config.nodeId}-", sharedServer.clientQuotaCallbackPlugin)
       clientQuotaMetadataManager = new ClientQuotaMetadataManager(quotaManagers, socketServer.connectionQuotas)
       controllerApis = new ControllerApis(socketServer.dataPlaneRequestChannel,
         authorizerPlugin,

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

@@ -21,6 +21,7 @@ import kafka.metrics.KafkaMetricsReporter
 import kafka.raft.KafkaRaftManager
 import kafka.server.Server.MetricsPrefix
 import kafka.utils.{Logging, VerifiableProperties}
+import org.apache.kafka.common.internals.Plugin
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.utils.{Time, Utils}
@@ -40,6 +41,7 @@ import org.apache.kafka.server.config.DefaultSupportedConfigChecker
 import org.apache.kafka.server.common.ApiMessageAndVersion
 import org.apache.kafka.server.fault.{FaultHandler, LoggingFaultHandler, ProcessTerminatingFaultHandler}
 import org.apache.kafka.server.metrics.{BrokerServerMetrics, KafkaYammerMetrics, NodeMetrics}
+import org.apache.kafka.server.quota.ClientQuotaCallback
 
 import java.net.InetSocketAddress
 import java.util.Arrays
@@ -114,6 +116,10 @@ class SharedServer(
   private var usedByController: Boolean = false
   val brokerConfig = new KafkaConfig(sharedServerConfig.props, false)
   val controllerConfig = new KafkaConfig(sharedServerConfig.props, false)
+  // Create the client quota callback once so that, in combined mode, the broker and controller share a
+  // single instance instead of each instantiating their own via reflection with divergent state (KAFKA-20650).
+  val clientQuotaCallbackPlugin: Optional[Plugin[ClientQuotaCallback]] =
+    QuotaFactory.createClientQuotaCallback(sharedServerConfig, _metrics, ProcessRole.BrokerRole.toString)
   val supportedConfigChecker: SupportedConfigChecker = new DefaultSupportedConfigChecker()
 
   // Factory for creating request handler pools with shared aggregate thread counter