This is an automated email from the ASF dual-hosted git repository. showuon pushed a commit to branch 3.6 in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.6 by this push: new c74a9832df4 KAFKA-16222: desanitize entity name when migrate client quotas (#15481) c74a9832df4 is described below commit c74a9832df474367da1feb3ee85f9f5e98a24968 Author: PoAn Yang <pay...@apache.org> AuthorDate: Wed Mar 20 14:53:23 2024 +0800 KAFKA-16222: desanitize entity name when migrate client quotas (#15481) The entity name is sanitized when it's in Zk mode. We didn't desanitize it when we migrate client quotas. Add Sanitizer.desanitize to fix it. Reviewers: Luke Chen <show...@gmail.com> --- .../main/scala/kafka/zk/ZkMigrationClient.scala | 4 +++ .../kafka/zk/ZkMigrationIntegrationTest.scala | 31 ++++++++++++++-------- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala index a11a84c017b..76e0b47aee8 100644 --- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala +++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala @@ -27,6 +27,7 @@ import org.apache.kafka.common.errors.ControllerMovedException import org.apache.kafka.common.metadata._ import org.apache.kafka.common.resource.ResourcePattern import org.apache.kafka.common.security.scram.ScramCredential +import org.apache.kafka.common.utils.Sanitizer import org.apache.kafka.common.{TopicIdPartition, Uuid} import org.apache.kafka.metadata.DelegationTokenData import org.apache.kafka.metadata.PartitionRegistration @@ -225,6 +226,9 @@ class ZkMigrationClient( entityDataList: util.List[ClientQuotaRecord.EntityData], quotas: util.Map[String, lang.Double] ): Unit = { + entityDataList.forEach(entityData => { + entityData.setEntityName(Sanitizer.desanitize(entityData.entityName())) + }) val batch = new util.ArrayList[ApiMessageAndVersion]() quotas.forEach((key, value) => { batch.add(new ApiMessageAndVersion(new ClientQuotaRecord() diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala index 4c3ddd8b80d..f8fe82ed9d0 100644 --- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala @@ -17,7 +17,7 @@ package kafka.zk import kafka.security.authorizer.AclEntry.{WildcardHost, WildcardPrincipalString} -import kafka.server.{ConfigType, ControllerRequestCompletionHandler, KafkaConfig} +import kafka.server.{ConfigEntityName, ConfigType, ControllerRequestCompletionHandler, KafkaConfig} import kafka.test.{ClusterConfig, ClusterGenerator, ClusterInstance} import kafka.test.annotation.{AutoStart, ClusterConfigProperty, ClusterTemplate, ClusterTest, Type} import kafka.test.junit.ClusterTestExtensions @@ -54,7 +54,6 @@ import org.slf4j.LoggerFactory import java.util import java.util.concurrent.{CompletableFuture, ExecutionException, TimeUnit} import java.util.{Collections, Optional, Properties, UUID} -import scala.collection.Seq import scala.jdk.CollectionConverters._ object ZkMigrationIntegrationTest { @@ -217,15 +216,19 @@ class ZkMigrationIntegrationTest { createTopicResult.all().get(60, TimeUnit.SECONDS) val quotas = new util.ArrayList[ClientQuotaAlteration]() - quotas.add(new ClientQuotaAlteration( - new ClientQuotaEntity(Map("user" -> "user1").asJava), - List(new ClientQuotaAlteration.Op("consumer_byte_rate", 1000.0)).asJava)) - quotas.add(new ClientQuotaAlteration( - new ClientQuotaEntity(Map("user" -> "user1", "client-id" -> "clientA").asJava), + val defaultUserEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.USER -> ConfigEntityName.Default).asJava) + quotas.add(new ClientQuotaAlteration(defaultUserEntity, List(new ClientQuotaAlteration.Op("consumer_byte_rate", 900.0)).asJava)) + val defaultClientIdEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> ConfigEntityName.Default).asJava) + quotas.add(new ClientQuotaAlteration(defaultClientIdEntity, List(new ClientQuotaAlteration.Op("consumer_byte_rate", 900.0)).asJava)) + val defaultIpEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> null.asInstanceOf[String]).asJava) + quotas.add(new ClientQuotaAlteration(defaultIpEntity, List(new ClientQuotaAlteration.Op("connection_creation_rate", 9.0)).asJava)) + val userEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.USER -> "user/1@prod").asJava) + quotas.add(new ClientQuotaAlteration(userEntity, List(new ClientQuotaAlteration.Op("consumer_byte_rate", 1000.0)).asJava)) + val userClientEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.USER -> "user/1@prod", ClientQuotaEntity.CLIENT_ID -> "client/1@domain").asJava) + quotas.add(new ClientQuotaAlteration(userClientEntity, List(new ClientQuotaAlteration.Op("consumer_byte_rate", 800.0), new ClientQuotaAlteration.Op("producer_byte_rate", 100.0)).asJava)) - quotas.add(new ClientQuotaAlteration( - new ClientQuotaEntity(Map("ip" -> "8.8.8.8").asJava), - List(new ClientQuotaAlteration.Op("connection_creation_rate", 10.0)).asJava)) + val ipEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "8.8.8.8").asJava) + quotas.add(new ClientQuotaAlteration(ipEntity, List(new ClientQuotaAlteration.Op("connection_creation_rate", 10.0)).asJava)) admin.alterClientQuotas(quotas).all().get(60, TimeUnit.SECONDS) val zkClient = clusterInstance.asInstanceOf[ZkClusterInstance].getUnderlying().zkClient @@ -261,7 +264,13 @@ class ZkMigrationIntegrationTest { assertEquals(10, image.topics().getTopic("test-topic-3").partitions().size()) val clientQuotas = image.clientQuotas().entities() - assertEquals(3, clientQuotas.size()) + assertEquals(6, clientQuotas.size()) + assertEquals(true, clientQuotas.containsKey(defaultUserEntity)) + assertEquals(true, clientQuotas.containsKey(defaultClientIdEntity)) + assertEquals(true, clientQuotas.containsKey(new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "").asJava))) // default ip + assertEquals(true, clientQuotas.containsKey(userEntity)) + assertEquals(true, clientQuotas.containsKey(userClientEntity)) + assertEquals(true, clientQuotas.containsKey(ipEntity)) } migrationState = migrationClient.releaseControllerLeadership(migrationState)