Broker support for QuotaRecord and DescribeClientQuotas by mumrah · Pull Request #477 · confluentinc/kafka

David Arthur (mumrah) · 2020-12-30T18:22:07Z

This PR adds support for handling QuotaRecord metadata in the broker. It also includes a new QuotaCache class for efficient lookups of the quotas.

* Add broker metadata processor for QuotaRecord-s * Add a QuotaCache for fast lookups * Integration with KafkaApis

David Arthur (mumrah) · 2020-12-30T18:23:06Z

          .setEntries(entriesData.asJava)))
    } else {
+      val result = quotaCache.describeClientQuotas(describeClientQuotasRequest.filter())
+      val resultAsJava = new util.HashMap[ClientQuotaEntity, util.Map[String, java.lang.Double]](result.size)


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

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

Ron Dagostino (rondagostino)

LGTM. Nothing major, just a few comments.

BTW, I'm seeing a spotbugs error on SocketServer.java:1458 (need to remove the trailing intValue() from QuotaConfigs.IP_CONNECTION_RATE_DEFAULT.intValue())

Ron Dagostino (rondagostino) · 2021-01-04T16:32:46Z

          .setThrottleTimeMs(requestThrottleMs)
          .setEntries(entriesData.asJava)))
    } else {
+      val result = quotaCache.describeClientQuotas(describeClientQuotasRequest.filter())


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?

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)

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.

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.

Ron Dagostino (rondagostino) · 2021-01-04T16:51:23Z

          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())


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.

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.

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?

Ron Dagostino (rondagostino) · 2021-01-04T16:57:23Z

          .setEntries(entriesData.asJava)))
    } else {
+      val result = quotaCache.describeClientQuotas(describeClientQuotasRequest.filter())
+      val resultAsJava = new util.HashMap[ClientQuotaEntity, util.Map[String, java.lang.Double]](result.size)


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

Ron Dagostino (rondagostino) · 2021-01-04T17:02:14Z

  private val replicaQuotaManager: ReplicationQuotaManager = EasyMock.createNiceMock(classOf[ReplicationQuotaManager])
  private val quotas = QuotaManagers(clientQuotaManager, clientQuotaManager, clientRequestQuotaManager,
    clientControllerQuotaManager, replicaQuotaManager, replicaQuotaManager, replicaQuotaManager, None)
+  private val quotaCache = new QuotaCache()


Interestingly, it appears we don't have a test for handleDescribeClientQuotasRequest() at all. Just noting the fact -- nothing to really do unless you want to tackle adding one as part of this PR.

Ron Dagostino (rondagostino) · 2021-01-04T17:19:56Z

+      case IpEntity(ip) => try {
+        Some(InetAddress.getByName(ip))
+      } catch {
+        case _: UnknownHostException => throw new IllegalArgumentException(s"Unable to resolve address $ip")


We warn about unknown types/keys elsewhere, so maybe we should warn in this case as well? Otherwise nothing else will get processed from this message.

Hmm, yea. It's actually a bit dubious to try and resolve the IP when the quota is set. What if there is a network issue at that moment? We'll miss the record and never update the quota manager.

Ron Dagostino (rondagostino) · 2021-01-04T17:48:08Z

+      // TODO return empty or throw exception here?
+      return Map.empty


Does this code execute if a client tries to describe everything -- and is that legal to do? I'm not sure.

The kafka-configs.sh script requires at least one entity type filter, so we can't reach this code via the CLI. However, a client could send an RPC message with no filters. Looks like the current ZK logic prohibits a describe without at least one entity type filter, so we'll do the same.

…a-processor-2 Conflicts: core/src/main/scala/kafka/server/Kip500Broker.scala

Ron Dagostino (rondagostino)

LGTM except for the outstanding question about throw new IllegalArgumentException(s"Unable to resolve address $ip") when processing an IP quota record. Maybe Colin Patrick McCabe (@cmccabe) has some thoughts on this.

Ron Dagostino (rondagostino) · 2021-01-09T01:15:35Z

+        DescribeClientQuotasResponse.fromQuotaEntities(resultAsJava, requestThrottleMs)
+      )
    } else {
+      warn("Neither LegacyAdminManager nor QuotaCache were defined")


ERROR level since this should never happen?

…a-processor-2

José Armando García Sancio (jsancio)

Thanks for the PR. Some high-level comments.

José Armando García Sancio (jsancio) · 2021-01-14T21:03:56Z

+      case DefaultIpEntity =>
+        updateCacheIndexPartial(ipEntityIndex, DefaultIp)
+    }
+  }


Hmm. I think you can simplify this code if you instead have a function convert that knows how to convert a QuotaEntity to a List[(QuotaCacheIndex, CacheIndexKey)]. With this you can do something like:

convert(entity).foreach { (quotaCacheIndex, cacheIndexKey) => updateCacheIndex(entity, removeFromIndex, quotaCacheIndex, cacheIndexKey) }

…a-processor-2 Conflicts: core/src/main/scala/kafka/server/KafkaApis.scala core/src/main/scala/kafka/server/Kip500Broker.scala core/src/main/scala/kafka/server/LegacyBroker.scala core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala core/src/test/scala/integration/kafka/server/Kip500ClusterTest.scala core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

Colin Patrick McCabe (cmccabe) · 2021-01-29T17:46:08Z

LGTM

Colin Patrick McCabe (cmccabe) · 2021-01-29T17:48:05Z

Like we discussed outside github, it's kind of unfortunate that we have several internal Scala types for client quotas. A public Java API type + a single internal Scala type should be enough, and that would cut down on the translations. But we should probably fix that later once 2.8 is out.

Thanks, David Arthur (@mumrah) !

I'll commit once the conflict is fixed

…a-processor-2

Broker support for QuotaRecord and DescribeClientQuotas

8ef5a1a

* Add broker metadata processor for QuotaRecord-s * Add a QuotaCache for fast lookups * Integration with KafkaApis

David Arthur (mumrah) commented Dec 30, 2020

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java Outdated

David Arthur (mumrah) commented Dec 30, 2020

View reviewed changes

Restructure the cache logic a bit

2a1e7d9

Ron Dagostino (rondagostino) approved these changes Jan 4, 2021

View reviewed changes

David Arthur (mumrah) added 5 commits January 4, 2021 15:56

Small bit of cleanup

1bdb82a

PR feedback

e3fa08f

Add some comments, cleanup a few things

13bb515

Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…

fc9cc3d

…a-processor-2 Conflicts: core/src/main/scala/kafka/server/Kip500Broker.scala

Fix imports

8ad11d4

Ron Dagostino (rondagostino) reviewed Jan 9, 2021

View reviewed changes

Colin Patrick McCabe (cmccabe) added the kip-500 label Jan 11, 2021

David Arthur (mumrah) added 3 commits January 12, 2021 15:20

Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…

b81b4c0

…a-processor-2

Add an end to end integration test for client quotas

25f96ae

Add more cases to the integration test

a07cceb

José Armando García Sancio (jsancio) reviewed Jan 14, 2021

View reviewed changes

David Arthur (mumrah) added 5 commits January 28, 2021 11:14

PR feedback

4503dc4

Indent fix

81c46f3

Remove unused arg

3efd9b9

Rename some things

d0390b0

Merge remote-tracking branch 'confluentinc/kip-500' into kip-500-quot…

0f3f8f3

…a-processor-2

Colin Patrick McCabe (cmccabe) merged commit 01b1e4d into kip-500 Jan 29, 2021

David Arthur (mumrah) mentioned this pull request Feb 10, 2021

Add ClientQuotaMetadataManager for processing QuotaRecord apache/kafka#10101

Merged

		// TODO return empty or throw exception here?
		return Map.empty

Uh oh!

Conversation

David Arthur (mumrah) commented Dec 30, 2020

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ron Dagostino (rondagostino) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ron Dagostino (rondagostino) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

José Armando García Sancio (jsancio) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Colin Patrick McCabe (cmccabe) commented Jan 29, 2021

Uh oh!

Colin Patrick McCabe (cmccabe) commented Jan 29, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants