diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java
index cf24d858e5865..e03287843c35d 100644
--- a/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java
+++ b/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java
@@ -97,6 +97,12 @@ default T deserialize(String topic, Headers headers, byte[] data) {
*
Similarly, if this method is overridden, the implementation cannot make any assumptions about the
* passed in {@link ByteBuffer} either.
*
+ *
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.
+ *
*
It is recommended to deserialize a {@code null} {@link ByteBuffer} to a {@code null} object.
*
*
Note that the passed in {@link Headers} may be empty, but never {@code null}.
diff --git a/core/src/main/java/kafka/server/QuotaFactory.java b/core/src/main/java/kafka/server/QuotaFactory.java
index aa28856cdbbff..a32fee7e41ee6 100644
--- a/core/src/main/java/kafka/server/QuotaFactory.java
+++ b/core/src/main/java/kafka/server/QuotaFactory.java
@@ -103,7 +103,31 @@ public static QuotaManagers instantiate(
);
}
- private static Optional> 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> 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> createClientQuotaCallback(
KafkaConfig cfg,
Metrics metrics,
String role
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index d7502aaa039e3..01f1bdde75315 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/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 */
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index 56f036f090f0f..c06d8fb2a11e5 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/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,
diff --git a/core/src/main/scala/kafka/server/SharedServer.scala b/core/src/main/scala/kafka/server/SharedServer.scala
index fcc39b94becd5..464d44886efd9 100644
--- a/core/src/main/scala/kafka/server/SharedServer.scala
+++ b/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