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 2 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
18 changes: 13 additions & 5 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: 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 @@ -3205,8 +3206,15 @@ class KafkaApis(val requestChannel: RequestChannel,
.setThrottleTimeMs(requestThrottleMs)
.setEntries(entriesData.asJava)))
} else {
val result = quotaCache.describeClientQuotas(describeClientQuotasRequest.filter())

@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.

I think we need an authorization check here. The above authorization check only occurs if we are in legacy mode. Maybe perform a single authorization check at the top before branching based on adminManager being null or not?

@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 also occurs to me that we need tests for these KIP-500 states (this comment is a general comment, so probably unnecessary to add the test now -- just something we'll have to do before opening a PR to the apache repo)

@mumrah David Arthur (mumrah) Jan 4, 2021

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.

Yea, thanks I missed the authz thing. Will fix.

What kind of tests are you thinking? I think KafkaApis is mainly tested through integration/system tests.

@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.

I could see a unit test that confirms unauthorized requests are denied (one version for legacy, one version for KIP-500). I could also see a KIP-500 unit test that confirms a mock QuotaCache is asked to handle the request. Then as long as QuotaCache is tested that should be enough.

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(new ClientQuotaEntity(entity.toMap.asJava),
quotas.map { case (key, quota) => key -> Double.box(quota)}.asJava)
}
apisUtils.sendResponseMaybeThrottle(request, requestThrottleMs =>
describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
DescribeClientQuotasResponse.fromQuotaEntities(resultAsJava, requestThrottleMs)
)
}
}

Expand Down
17 changes: 12 additions & 5 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 @@ -20,7 +20,6 @@ package kafka.server
import java.util.concurrent.CompletableFuture
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock

import kafka.cluster.{Broker, EndPoint}
import kafka.controller.KafkaController
import kafka.coordinator.group.GroupCoordinator
Expand All @@ -30,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.feature.{Features, SupportedVersionRange}
Expand Down Expand Up @@ -107,8 +106,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 @@ -161,6 +163,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 @@ -216,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()

lifecycleManager.start(() => brokerMetadataListener.currentMetadataOffset(),
Expand Down Expand Up @@ -277,7 +282,8 @@ class Kip500Broker(
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, brokerMetadataListener)
fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache, brokerMetadataListener,
quotaCache)

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

controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time,
1, s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.ControlPlaneThreadPrefix)
Expand Down
8 changes: 5 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 All @@ -33,6 +32,7 @@ import kafka.log.LogManager
import kafka.metrics.{KafkaMetricsReporter, KafkaYammerMetrics}
import kafka.network.SocketServer
import kafka.security.CredentialProvider
import kafka.server.metadata.QuotaCache
import kafka.utils._
import kafka.zk.{BrokerInfo, KafkaZkClient}
import org.apache.kafka.clients.{ApiVersions, ClientDnsLookup, ManualMetadataUpdater, NetworkClient, NetworkClientUtils}
Expand Down Expand Up @@ -326,7 +326,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,
new QuotaCache())

@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.

Should this be null since it isn't used in the legacy case? Not sure if we should change the parameter to be an Option[] and default it to None so we don't have to pass it here? Although we don't use Option[] for brokerMetadataListener and we pass in null here for that, so probably not.

@mumrah David Arthur (mumrah) Jan 4, 2021

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.

Yea we could use an Option here instead of an empty cache that never gets used. I'll change that.

As an aside: it's going to get pretty messy in this class as we add members which are optional (or nullable). Maybe we should try to extract an interface from LegacyAdminManager and use that rather than adding more members to KafkaApis for kip-500 functionality.

@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.

I think at some point the ZooKeeper-related stuff will be removed -- that might be a good time to try to trim things down/clean it up a bit?


dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)
Expand All @@ -335,7 +336,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,
new QuotaCache())
Comment thread
mumrah marked this conversation as resolved.
Outdated

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 @@ -47,13 +48,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