Skip to content

Broker support for QuotaRecord and DescribeClientQuotas #477

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Colin Patrick McCabe (cmccabe) merged 16 commits into from
Jan 29, 2021
Merged

Broker support for QuotaRecord and DescribeClientQuotas #477

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
8ef5a1a
Broker support for QuotaRecord and DescribeClientQuotas
mumrah Dec 30, 2020
2a1e7d9
Restructure the cache logic a bit
mumrah Dec 30, 2020
1bdb82a
Small bit of cleanup
mumrah Jan 4, 2021
e3fa08f
PR feedback
mumrah Jan 4, 2021
13bb515
Add some comments, cleanup a few things
mumrah Jan 5, 2021
fc9cc3d
Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…
mumrah Jan 8, 2021
8ad11d4
Fix imports
mumrah Jan 8, 2021
b81b4c0
Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…
mumrah Jan 12, 2021
25f96ae
Add an end to end integration test for client quotas
mumrah Jan 12, 2021
a07cceb
Add more cases to the integration test
mumrah Jan 13, 2021
4503dc4
PR feedback
mumrah Jan 28, 2021
b441431
Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…
mumrah Jan 28, 2021
81c46f3
Indent fix
mumrah Jan 28, 2021
3efd9b9
Remove unused arg
mumrah Jan 28, 2021
d0390b0
Rename some things
mumrah Jan 28, 2021
0f3f8f3
Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…
mumrah Jan 29, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 22 additions & 6 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import java.util
import java.util.{Collections, Optional, Properties}
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

import kafka.admin.{AdminUtils, RackAwareMode}
import kafka.api.ElectLeadersRequestOps
import kafka.api.{ApiVersion, KAFKA_0_11_0_IV0, KAFKA_2_3_IV0}
Expand Down Expand Up @@ -90,8 +89,9 @@ import scala.collection.mutable.ArrayBuffer
import scala.collection.{Map, Seq, Set, immutable, mutable}
import scala.util.{Failure, Success, Try}
import kafka.coordinator.group.GroupOverview
import kafka.server.metadata.BrokerMetadataListener
import kafka.server.metadata.{BrokerMetadataListener, QuotaCache}
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.quota.ClientQuotaEntity
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource

import scala.annotation.nowarn
Expand Down Expand Up @@ -120,7 +120,8 @@ class KafkaApis(val requestChannel: RequestChannel,
val tokenManager: DelegationTokenManager,
val brokerFeatures: BrokerFeatures,
val finalizedFeatureCache: FinalizedFeatureCache,
brokerMetadataListener: BrokerMetadataListener) extends ApiRequestHandler with Logging {
brokerMetadataListener: BrokerMetadataListener,
val quotaCache: Option[QuotaCache]) extends ApiRequestHandler with Logging {

val apisUtils = new ApisUtils(new LogContext(s"[BrokerApis id=${config.brokerId}] "),
requestChannel, authorizer, quotas, time, Some(groupCoordinator), Some(txnCoordinator))
Expand Down Expand Up @@ -249,7 +250,7 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.ALTER_PARTITION_REASSIGNMENTS => maybeForward(request, handleAlterPartitionReassignmentsRequest)
case ApiKeys.LIST_PARTITION_REASSIGNMENTS => handleListPartitionReassignmentsRequest(request)
case ApiKeys.OFFSET_DELETE => handleOffsetDeleteRequest(request)
case ApiKeys.DESCRIBE_CLIENT_QUOTAS => maybeForward(request, handleDescribeClientQuotasRequest)
case ApiKeys.DESCRIBE_CLIENT_QUOTAS => handleDescribeClientQuotasRequest(request)
case ApiKeys.ALTER_CLIENT_QUOTAS => maybeForward(request, handleAlterClientQuotasRequest)
case ApiKeys.DESCRIBE_USER_SCRAM_CREDENTIALS => handleDescribeUserScramCredentialsRequest(request)
case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => maybeForward(request, handleAlterUserScramCredentialsRequest)
Expand Down Expand Up @@ -3180,7 +3181,10 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleDescribeClientQuotasRequest(request: RequestChannel.Request): Unit = {
val describeClientQuotasRequest = request.body[DescribeClientQuotasRequest]

if (adminManager != null && apisUtils.authorize(request.context, DESCRIBE_CONFIGS, CLUSTER, CLUSTER_NAME)) {
if (!apisUtils.authorize(request.context, DESCRIBE_CONFIGS, CLUSTER, CLUSTER_NAME)) {
apisUtils.sendResponseMaybeThrottle(request, requestThrottleMs =>
describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
} else if (adminManager != null) {
val result = adminManager.describeClientQuotas(describeClientQuotasRequest.filter)

val entriesData = result.iterator.map { case (quotaEntity, quotaValues) =>
Expand All @@ -3205,9 +3209,21 @@ class KafkaApis(val requestChannel: RequestChannel,
new DescribeClientQuotasResponse(new DescribeClientQuotasResponseData()
.setThrottleTimeMs(requestThrottleMs)
.setEntries(entriesData.asJava)))
} else if (quotaCache.isDefined) {
val result = quotaCache.get.describeClientQuotas(
describeClientQuotasRequest.filter().components().asScala.toSeq,
describeClientQuotasRequest.filter().strict())
val resultAsJava = new util.HashMap[ClientQuotaEntity, util.Map[String, java.lang.Double]](result.size)

@mumrah David Arthur (mumrah) Dec 30, 2020

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These conversions are annoying, but I don't know how to avoid them (unless we just use Java classes in the QuotaCache code)

@rondagostino Ron Dagostino (rondagostino) Jan 4, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not a hot path, so I personally don't see a real need to change the QuotaCache class -- though others may differ.

result.foreach { case (entity, quotas) =>
resultAsJava.put(entity, quotas.map { case (key, quota) => key -> Double.box(quota)}.asJava)
}
apisUtils.sendResponseMaybeThrottle(request, requestThrottleMs =>
DescribeClientQuotasResponse.fromQuotaEntities(resultAsJava, requestThrottleMs)
)
} else {
warn("Neither LegacyAdminManager nor QuotaCache were defined")

@rondagostino Ron Dagostino (rondagostino) Jan 9, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ERROR level since this should never happen?

apisUtils.sendResponseMaybeThrottle(request, requestThrottleMs =>
describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.UNKNOWN_SERVER_ERROR.exception))
}
}

Expand Down
14 changes: 10 additions & 4 deletions core/src/main/scala/kafka/server/Kip500Broker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import kafka.metrics.KafkaYammerMetrics
import kafka.network.SocketServer
import kafka.security.CredentialProvider
import kafka.server.KafkaBroker.metricsPrefix
import kafka.server.metadata.BrokerMetadataListener
import kafka.server.metadata.{BrokerMetadataListener, QuotaCache}
import kafka.utils.{CoreUtils, KafkaScheduler}
import kafka.zk.KafkaZkClient
import org.apache.kafka.common.internals.Topic
Expand Down Expand Up @@ -107,8 +107,11 @@ class Kip500Broker(
var kafkaScheduler: KafkaScheduler = null

var metadataCache: MetadataCache = null

var quotaManagers: QuotaFactory.QuotaManagers = null

var quotaCache: QuotaCache = null

private var _brokerTopicStats: BrokerTopicStats = null

val brokerFeatures: BrokerFeatures = BrokerFeatures.createDefault()
Expand Down Expand Up @@ -158,6 +161,7 @@ class Kip500Broker(
_brokerTopicStats = new BrokerTopicStats

quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))
quotaCache = new QuotaCache()

logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)

Expand Down Expand Up @@ -215,11 +219,13 @@ class Kip500Broker(
/* Add all reconfigurables for config change notification before starting the metadata listener */
config.dynamicConfig.addReconfigurables(this)



Comment thread
mumrah marked this conversation as resolved.
brokerMetadataListener = new BrokerMetadataListener(
config, metadataCache, time,
BrokerMetadataListener.defaultProcessors(
config, clusterId, metadataCache, groupCoordinator, quotaManagers, replicaManager, transactionCoordinator,
logManager))
logManager, socketServer, quotaCache))
brokerMetadataListener.start()

val networkListeners = new ListenerCollection()
Expand Down Expand Up @@ -280,7 +286,7 @@ class Kip500Broker(
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel,
replicaManager, null, groupCoordinator, transactionCoordinator,
null, forwardingManager, null, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, brokerMetadataListener)
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, brokerMetadataListener, Some(quotaCache))

dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)
Expand All @@ -289,7 +295,7 @@ class Kip500Broker(
controlPlaneRequestProcessor = new KafkaApis(controlPlaneRequestChannel,
replicaManager, null, groupCoordinator, transactionCoordinator,
null, forwardingManager, null, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, brokerMetadataListener)
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, brokerMetadataListener, Some(quotaCache))

controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time,
1, s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.ControlPlaneThreadPrefix)
Expand Down
7 changes: 4 additions & 3 deletions core/src/main/scala/kafka/server/LegacyBroker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import java.net.{InetAddress, SocketTimeoutException}
import java.util
import java.util.concurrent._
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger, AtomicReference}

import kafka.api.{KAFKA_0_9_0, KAFKA_2_2_IV0, KAFKA_2_4_IV1}
import kafka.cluster.Broker
import kafka.common.{GenerateBrokerIdException, InconsistentBrokerIdException, InconsistentClusterIdException}
Expand Down Expand Up @@ -332,7 +331,8 @@ class LegacyBroker(val config: KafkaConfig,
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel,
replicaManager, adminManager, groupCoordinator, transactionCoordinator,
kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
fetchManager, brokerTopicStats, _clusterId, time, tokenManager, brokerFeatures, featureCache, null)
fetchManager, brokerTopicStats, _clusterId, time, tokenManager, brokerFeatures, featureCache, null,
None)

dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)
Expand All @@ -341,7 +341,8 @@ class LegacyBroker(val config: KafkaConfig,
controlPlaneRequestProcessor = new KafkaApis(controlPlaneRequestChannel,
replicaManager, adminManager, groupCoordinator, transactionCoordinator,
kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
fetchManager, brokerTopicStats, _clusterId, time, tokenManager, brokerFeatures, featureCache, null)
fetchManager, brokerTopicStats, _clusterId, time, tokenManager, brokerFeatures, featureCache, null,
None)

controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time,
1, s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.ControlPlaneThreadPrefix)
Expand Down
8 changes: 6 additions & 2 deletions core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import kafka.coordinator.group.GroupCoordinator
import kafka.coordinator.transaction.TransactionCoordinator
import kafka.log.LogManager
import kafka.metrics.KafkaMetricsGroup
import kafka.network.SocketServer
import kafka.server._
import kafka.utils.ShutdownableThread
import org.apache.kafka.common.config.ConfigResource
Expand All @@ -46,13 +47,16 @@ object BrokerMetadataListener {
quotaManagers: QuotaFactory.QuotaManagers,
replicaManager: ReplicaManager,
txnCoordinator: TransactionCoordinator,
logManager: LogManager): List[BrokerMetadataProcessor] = {
logManager: LogManager,
socketServer: SocketServer,
quotaCache: QuotaCache): List[BrokerMetadataProcessor] = {
val configHandlers = Map[ConfigResource.Type, ConfigHandler](
ConfigResource.Type.TOPIC -> new TopicConfigHandler(logManager, kafkaConfig, quotaManagers, None),
ConfigResource.Type.BROKER -> new BrokerConfigHandler(kafkaConfig, quotaManagers))
List(
new PartitionMetadataProcessor(kafkaConfig, clusterId, metadataCache, groupCoordinator, quotaManagers,
replicaManager, txnCoordinator, configHandlers)
replicaManager, txnCoordinator, configHandlers),
new QuotaMetadataProcessor(quotaManagers, socketServer.connectionQuotas, quotaCache)
)
}
}
Expand Down
Loading