You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2024/03/26 00:04:14 UTC
(kafka) branch KAFKA-16411 created (now bf1900e9971)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a change to branch KAFKA-16411
in repository https://gitbox.apache.org/repos/asf/kafka.git
at bf1900e9971 Improve tests
This branch includes the following new commits:
new cbe5ada027c Rename 'mangling' to 'sanitization' to be more consistent
new bf1900e9971 Improve tests
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(kafka) 01/02: Rename 'mangling' to 'sanitization' to be more consistent
Posted by cm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch KAFKA-16411
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit cbe5ada027c1882b74ea55e998325f62be2f611b
Author: Colin P. McCabe <cm...@apache.org>
AuthorDate: Mon Mar 25 12:23:08 2024 -0700
Rename 'mangling' to 'sanitization' to be more consistent
---
.../kafka/zk/migration/ZkConfigMigrationClient.scala | 18 +++++++++---------
1 file changed, 9 insertions(+), 9 deletions(-)
diff --git a/core/src/main/scala/kafka/zk/migration/ZkConfigMigrationClient.scala b/core/src/main/scala/kafka/zk/migration/ZkConfigMigrationClient.scala
index 04a58894ae7..fda6e93c249 100644
--- a/core/src/main/scala/kafka/zk/migration/ZkConfigMigrationClient.scala
+++ b/core/src/main/scala/kafka/zk/migration/ZkConfigMigrationClient.scala
@@ -21,7 +21,7 @@ import kafka.server.{DynamicBrokerConfig, DynamicConfig, ZkAdminManager}
import kafka.utils.Logging
import kafka.zk.ZkMigrationClient.{logAndRethrow, wrapZkException}
import kafka.zk._
-import kafka.zk.migration.ZkConfigMigrationClient.getMangledClientQuotaZNodeName
+import kafka.zk.migration.ZkConfigMigrationClient.getSanitizedClientQuotaZNodeName
import kafka.zookeeper.{CreateRequest, DeleteRequest, SetDataRequest}
import org.apache.kafka.clients.admin.ScramMechanism
import org.apache.kafka.common.config.types.Password
@@ -81,8 +81,8 @@ class ZkConfigMigrationClient(
// which have their names set to the empty string instead.
result.setEntityName(null)
} else {
- // ZNode names are mangled before being stored in ZooKeeper.
- // For example, @ is turned into %40. Undo the mangling here.
+ // ZNode names are sanitized before being stored in ZooKeeper.
+ // For example, @ is turned into %40. Undo the sanitization here.
result.setEntityName(Sanitizer.desanitize(znodeName))
}
result
@@ -261,9 +261,9 @@ class ZkConfigMigrationClient(
scram: util.Map[String, String],
state: ZkMigrationLeadershipState
): ZkMigrationLeadershipState = wrapZkException {
- val user: Option[String] = getMangledClientQuotaZNodeName(entity, ClientQuotaEntity.USER)
- val client: Option[String] = getMangledClientQuotaZNodeName(entity, ClientQuotaEntity.CLIENT_ID)
- val ip: Option[String] = getMangledClientQuotaZNodeName(entity, ClientQuotaEntity.IP)
+ val user: Option[String] = getSanitizedClientQuotaZNodeName(entity, ClientQuotaEntity.USER)
+ val client: Option[String] = getSanitizedClientQuotaZNodeName(entity, ClientQuotaEntity.CLIENT_ID)
+ val ip: Option[String] = getSanitizedClientQuotaZNodeName(entity, ClientQuotaEntity.IP)
val props = new Properties()
val (configType, path, configKeys) = if (user.isDefined && client.isEmpty) {
@@ -369,7 +369,7 @@ object ZkConfigMigrationClient {
* @param component The component that we want a znode name for.
* @return Some(znodeName) if there is a znode path; None otherwise.
*/
- def getMangledClientQuotaZNodeName(
+ def getSanitizedClientQuotaZNodeName(
entity: util.Map[String, String],
component: String
): Option[String] = {
@@ -385,10 +385,10 @@ object ZkConfigMigrationClient {
// "not present." This is an unfortunate API that should be revisited at some point.
Some(ZooKeeperInternals.DEFAULT_STRING)
} else {
- // We found a non-null value, and now we need to mangle it. For example, "c@@ldude" will
+ // We found a non-null value, and now we need to sanitize it. For example, "c@@ldude" will
// turn into c%40%40ldude, so that we can use it as a znode name in ZooKeeper.
Some(Sanitizer.sanitize(rawValue))
}
}
}
-}
\ No newline at end of file
+}
(kafka) 02/02: Improve tests
Posted by cm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch KAFKA-16411
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit bf1900e9971cddd6c08e38baf40c9ae04dfc84cc
Author: Colin P. McCabe <cm...@apache.org>
AuthorDate: Mon Mar 25 17:01:37 2024 -0700
Improve tests
---
.../kafka/zk/ZkMigrationIntegrationTest.scala | 22 +++++++++++++---------
1 file changed, 13 insertions(+), 9 deletions(-)
diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
index 16d68b3ae38..a3e5ecc0814 100644
--- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
@@ -40,7 +40,7 @@ import org.apache.kafka.common.resource.ResourcePattern
import org.apache.kafka.common.resource.ResourceType.TOPIC
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.scram.internals.ScramCredentialUtils
-import org.apache.kafka.common.utils.SecurityUtils
+import org.apache.kafka.common.utils.{Sanitizer, SecurityUtils}
import org.apache.kafka.image.{MetadataDelta, MetadataImage, MetadataProvenance}
import org.apache.kafka.metadata.authorizer.StandardAcl
import org.apache.kafka.metadata.migration.ZkMigrationLeadershipState
@@ -881,11 +881,14 @@ class ZkMigrationIntegrationTest {
def alterClientQuotas(admin: Admin): AlterClientQuotasResult = {
val quotas = new util.ArrayList[ClientQuotaAlteration]()
quotas.add(new ClientQuotaAlteration(
- new ClientQuotaEntity(Map("user" -> "user1").asJava),
+ new ClientQuotaEntity(Map("user" -> "user@1").asJava),
List(new ClientQuotaAlteration.Op("consumer_byte_rate", 1000.0)).asJava))
quotas.add(new ClientQuotaAlteration(
- new ClientQuotaEntity(Map("user" -> "user1", "client-id" -> "clientA").asJava),
+ new ClientQuotaEntity(Map("user" -> "user@1", "client-id" -> "clientA").asJava),
List(new ClientQuotaAlteration.Op("consumer_byte_rate", 800.0), new ClientQuotaAlteration.Op("producer_byte_rate", 100.0)).asJava))
+ quotas.add(new ClientQuotaAlteration(
+ new ClientQuotaEntity(Collections.singletonMap("user", null)),
+ List(new ClientQuotaAlteration.Op("consumer_byte_rate", 900.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))
@@ -903,7 +906,7 @@ class ZkMigrationIntegrationTest {
val alterations = new util.ArrayList[UserScramCredentialAlteration]()
alterations.add(new UserScramCredentialUpsertion("user1",
new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_256, 8191), "password1"))
- alterations.add(new UserScramCredentialUpsertion("user2",
+ alterations.add(new UserScramCredentialUpsertion("user@2",
new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_256, 8192), "password2"))
admin.alterUserScramCredentials(alterations)
}
@@ -918,20 +921,21 @@ class ZkMigrationIntegrationTest {
def verifyClientQuotas(zkClient: KafkaZkClient): Unit = {
TestUtils.retry(10000) {
- assertEquals("1000", zkClient.getEntityConfigs(ConfigType.USER, "user1").getProperty("consumer_byte_rate"))
- assertEquals("800", zkClient.getEntityConfigs("users/user1/clients", "clientA").getProperty("consumer_byte_rate"))
- assertEquals("100", zkClient.getEntityConfigs("users/user1/clients", "clientA").getProperty("producer_byte_rate"))
+ assertEquals("1000", zkClient.getEntityConfigs(ConfigType.USER, Sanitizer.sanitize("user@1")).getProperty("consumer_byte_rate"))
+ assertEquals("900", zkClient.getEntityConfigs(ConfigType.USER, "<default>").getProperty("consumer_byte_rate"))
+ assertEquals("800", zkClient.getEntityConfigs("users/" + Sanitizer.sanitize("user@1") + "/clients", "clientA").getProperty("consumer_byte_rate"))
+ assertEquals("100", zkClient.getEntityConfigs("users/" + Sanitizer.sanitize("user@1") + "/clients", "clientA").getProperty("producer_byte_rate"))
assertEquals("10", zkClient.getEntityConfigs(ConfigType.IP, "8.8.8.8").getProperty("connection_creation_rate"))
}
}
def verifyUserScramCredentials(zkClient: KafkaZkClient): Unit = {
TestUtils.retry(10000) {
- val propertyValue1 = zkClient.getEntityConfigs(ConfigType.USER, "user1").getProperty("SCRAM-SHA-256")
+ val propertyValue1 = zkClient.getEntityConfigs(ConfigType.USER, Sanitizer.sanitize("user1")).getProperty("SCRAM-SHA-256")
val scramCredentials1 = ScramCredentialUtils.credentialFromString(propertyValue1)
assertEquals(8191, scramCredentials1.iterations)
- val propertyValue2 = zkClient.getEntityConfigs(ConfigType.USER, "user2").getProperty("SCRAM-SHA-256")
+ val propertyValue2 = zkClient.getEntityConfigs(ConfigType.USER, Sanitizer.sanitize("user@2")).getProperty("SCRAM-SHA-256")
assertNotNull(propertyValue2)
val scramCredentials2 = ScramCredentialUtils.credentialFromString(propertyValue2)
assertEquals(8192, scramCredentials2.iterations)