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 all 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
17 changes: 11 additions & 6 deletions core/src/main/scala/kafka/server/BrokerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,14 @@ import java.util
import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock

import kafka.coordinator.group.GroupCoordinator
import kafka.coordinator.transaction.{ProducerIdGenerator, TransactionCoordinator}
import kafka.log.LogManager
import kafka.metrics.KafkaYammerMetrics
import kafka.network.SocketServer
import kafka.security.CredentialProvider
import kafka.server.KafkaBroker.metricsPrefix
import kafka.server.metadata.{BrokerMetadataListener, LocalConfigRepository}
import kafka.server.metadata.{BrokerMetadataListener, LocalConfigRepository, ClientQuotaCache, ClientQuotaMetadataManager}
import kafka.utils.{CoreUtils, KafkaScheduler}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener, ListenerCollection}
Expand Down Expand Up @@ -107,8 +106,11 @@ class BrokerServer(
var kafkaScheduler: KafkaScheduler = null

var metadataCache: MetadataCache = null

var quotaManagers: QuotaFactory.QuotaManagers = null

var quotaCache: ClientQuotaCache = null

private var _brokerTopicStats: BrokerTopicStats = null

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

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

logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)

Expand Down Expand Up @@ -222,17 +225,19 @@ class BrokerServer(
/* Add all reconfigurables for config change notification before starting the metadata listener */
config.dynamicConfig.addReconfigurables(this)

val clientQuotaMetadataManager = new ClientQuotaMetadataManager(quotaManagers, socketServer.connectionQuotas, quotaCache)

brokerMetadataListener = new BrokerMetadataListener(
config.brokerId,
time,
metadataCache,
configRepository,
groupCoordinator,
quotaManagers,
replicaManager,
transactionCoordinator,
logManager,
threadNamePrefix)
threadNamePrefix,
clientQuotaMetadataManager)

val networkListeners = new ListenerCollection()
config.advertisedListeners.foreach { ep =>
Expand Down Expand Up @@ -292,7 +297,7 @@ class BrokerServer(
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, configRepository)
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, configRepository, Some(quotaCache))

dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)
Expand All @@ -301,7 +306,7 @@ class BrokerServer(
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, configRepository)
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, configRepository, Some(quotaCache))

controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time,
1, s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.ControlPlaneThreadPrefix)
Expand Down
27 changes: 23 additions & 4 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 @@ -87,9 +87,10 @@ 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.ConfigRepository
import kafka.server.metadata.{ConfigRepository, ClientQuotaCache}
import org.apache.kafka.clients.ClientResponse
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 @@ -118,7 +119,8 @@ class KafkaApis(val requestChannel: RequestChannel,
val tokenManager: DelegationTokenManager,
val brokerFeatures: BrokerFeatures,
val finalizedFeatureCache: FinalizedFeatureCache,
val configRepository: ConfigRepository) extends ApiRequestHandler with Logging {
val configRepository: ConfigRepository,
val quotaCache: Option[ClientQuotaCache]) extends ApiRequestHandler with Logging {

type FetchResponseStats = Map[TopicPartition, RecordConversionStats]
this.logIdent = "[KafkaApi-%d] ".format(brokerId)
Expand Down Expand Up @@ -247,7 +249,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 @@ -3201,7 +3203,10 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleDescribeClientQuotasRequest(request: RequestChannel.Request): Unit = {
val describeClientQuotasRequest = request.body[DescribeClientQuotasRequest]

if (adminManager != null && authHelper.authorize(request.context, DESCRIBE_CONFIGS, CLUSTER, CLUSTER_NAME)) {
if (!authHelper.authorize(request.context, DESCRIBE_CONFIGS, CLUSTER, CLUSTER_NAME)) {
requestHelper.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 @@ -3226,9 +3231,23 @@ 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)
}
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
DescribeClientQuotasResponse.fromQuotaEntities(resultAsJava, requestThrottleMs)
)
} else {
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
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?

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

Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/server/KafkaServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,7 @@ class KafkaServer(
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 @@ -342,7 +342,7 @@ class KafkaServer(
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
14 changes: 11 additions & 3 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 @@ -22,7 +22,7 @@ import kafka.coordinator.group.GroupCoordinator
import kafka.coordinator.transaction.TransactionCoordinator
import kafka.log.LogManager
import kafka.metrics.KafkaMetricsGroup
import kafka.server.{MetadataCache, QuotaFactory, ReplicaManager, RequestHandlerHelper}
import kafka.server.{MetadataCache, ReplicaManager, RequestHandlerHelper}
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.metadata.MetadataRecordType._
import org.apache.kafka.common.metadata._
Expand All @@ -42,11 +42,11 @@ class BrokerMetadataListener(val brokerId: Int,
val metadataCache: MetadataCache,
val configRepository: LocalConfigRepository,
val groupCoordinator: GroupCoordinator,
val quotaManagers: QuotaFactory.QuotaManagers,
val replicaManager: ReplicaManager,
val txnCoordinator: TransactionCoordinator,
val logManager: LogManager,
val threadNamePrefix: Option[String]
val threadNamePrefix: Option[String],
val clientQuotaManager: ClientQuotaMetadataManager
) extends MetaLogListener with KafkaMetricsGroup {
val logContext = new LogContext(s"[BrokerMetadataListener id=${brokerId}] ")
val log = logContext.logger(classOf[BrokerMetadataListener])
Expand Down Expand Up @@ -159,6 +159,8 @@ class BrokerMetadataListener(val brokerId: Int,
record.asInstanceOf[UnfenceBrokerRecord])
case REMOVE_TOPIC_RECORD => handleRemoveTopicRecord(imageBuilder,
record.asInstanceOf[RemoveTopicRecord])
case QUOTA_RECORD => handleQuotaRecord(imageBuilder,
record.asInstanceOf[QuotaRecord])
// TODO: handle FEATURE_LEVEL_RECORD
case _ => throw new RuntimeException(s"Unsupported record type ${recordType}")
}
Expand Down Expand Up @@ -224,6 +226,12 @@ class BrokerMetadataListener(val brokerId: Int,
groupCoordinator.handleDeletedPartitions(removedPartitions.map(_.toTopicPartition()).toSeq)
}

def handleQuotaRecord(imageBuilder: MetadataImageBuilder,
record: QuotaRecord): Unit = {
// TODO add quotas to MetadataImageBuilder
clientQuotaManager.handleQuotaRecord(record)
}

class HandleNewLeaderEvent(leader: MetaLogLeader)
extends EventQueue.FailureLoggingEvent(log) {
override def run(): Unit = {
Expand Down
Loading