You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2025/01/20 13:34:53 UTC
[PR] KAFKA 18530 Rename ZooKeeperInternals [kafka]
m1a2st opened a new pull request, #18641:
URL: https://github.com/apache/kafka/pull/18641
Jira: https://issues.apache.org/jira/browse/KAFKA-18530
Since Kafka 4.0 has deprecated ZooKeeper, we should also remove the `ZooKeeperInternals` class. However, we need to maintain backward compatibility for metric names.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1937287973
##########
core/src/main/scala/kafka/server/ClientQuotaManager.scala:
##########
@@ -451,10 +449,31 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
}
}
+ private def getOrDefaultClient(
+ sanitizedClientId: Option[ClientQuotaEntity.ConfigEntity]
+ ): Option[ClientQuotaEntity.ConfigEntity] = {
+ if (sanitizedClientId.isEmpty)
+ None
+ else if (sanitizedClientId.get.name() == DefaultString)
Review Comment:
in kraft, we don't use `<default>` so this check is weird. If this is used to fix test, then maybe we should revise the test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #18641:
URL: https://github.com/apache/kafka/pull/18641
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1944148844
##########
core/src/main/scala/kafka/server/ClientQuotaManager.scala:
##########
@@ -418,14 +421,11 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
*/
lock.writeLock().lock()
try {
- val userEntity = sanitizedUser.map {
- case ZooKeeperInternals.DEFAULT_STRING => DefaultUserEntity
- case user => UserEntity(user)
- }
- val clientIdEntity = sanitizedClientId.map {
- case ZooKeeperInternals.DEFAULT_STRING => DefaultClientIdEntity
- case _ => ClientIdEntity(clientId.getOrElse(throw new IllegalStateException("Client-id not provided")))
+ val clientIdEntity = clientEntity match {
Review Comment:
we don't need this conversion, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1928473563
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -147,13 +148,13 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
// Convert entity into Options with sanitized values for QuotaManagers
val (sanitizedUser, sanitizedClientId) = quotaEntity match {
case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
+ case DefaultUserEntity => (Some(ClientQuotaManager.DefaultString), None)
Review Comment:
We should use Option[BaseUserEntity] instead of a string. As mentioned previously (https://github.com/apache/kafka/pull/18641#discussion_r1922770484), there's no need for a specific string to represent the "default" value.
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -146,9 +146,7 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
private val quotaManagers: QuotaManagers) extends ConfigHandler with Logging {
def processConfigChanges(brokerId: String, properties: Properties): Unit = {
- if (brokerId == ZooKeeperInternals.DEFAULT_STRING)
- brokerConfig.dynamicConfig.updateDefaultConfig(properties)
- else if (brokerConfig.brokerId == brokerId.trim.toInt) {
+ if (brokerConfig.brokerId == brokerId.trim.toInt) {
Review Comment:
We must handle the "default" configs when the broker id is "empty"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18641:
URL: https://github.com/apache/kafka/pull/18641#issuecomment-2639324365
@m1a2st please file a PR for 4.0 as there are some conflicts
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #18641:
URL: https://github.com/apache/kafka/pull/18641#issuecomment-2630751397
Thanks for @chia7712 review, addressed all comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18641:
URL: https://github.com/apache/kafka/pull/18641#issuecomment-2639497402
> please file a PR for 4.0 as there are some conflicts
I backport some cleanup to eliminate the conflicts
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1922566817
##########
server-common/src/main/java/org/apache/kafka/server/config/KraftInternals.java:
##########
@@ -16,12 +16,11 @@
*/
package org.apache.kafka.server.config;
-public class ZooKeeperInternals {
+public class KraftInternals {
Review Comment:
Removing this would affect the default username and default client-id. Since this is a breaking change, I suggest we maintain consistency by keeping these default values. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1942960762
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -145,27 +149,42 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, sanitizedClientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ sanitizedClientEntity = sanitizedClientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ quotaEntity match {
+ case UserEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
+ case DefaultUserEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), None)
+ case ClientIdEntity(clientId) =>
+ (None, Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
Review Comment:
Do we need to sanitize data in kraft mode? the zk path is gone so we don't need to align the data for zk anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1938724850
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
in kraft mode, the two levels are still existent. could you please revise it instead of removing it?
##########
core/src/main/scala/kafka/server/DynamicConfig.scala:
##########
@@ -85,12 +85,10 @@ object DynamicConfig {
def validate(props: Properties): util.Map[String, AnyRef] = DynamicConfig.validate(ipConfigs, props, customPropsAllowed = false)
def isValidIpEntity(ip: String): Boolean = {
Review Comment:
this is used by `ConfigCommand` only, so could you please move this method to `ConfigCommand`?
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -145,27 +149,43 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, sanitizedClientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ sanitizedClientEntity = sanitizedClientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ val (userEntity, sanitizedClientEntity) = quotaEntity match {
Review Comment:
it is unnecessary to declare `val (userEntity, sanitizedClientEntity)`
```scala
quotaEntity match {
case UserEntity(user) =>
(Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
case DefaultUserEntity =>
(Some(ClientQuotaManager.DefaultUserEntity), None)
case ClientIdEntity(clientId) =>
(None, Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
case DefaultClientIdEntity =>
(None, Some(ClientQuotaManager.DefaultClientIdEntity))
case ExplicitUserExplicitClientIdEntity(user, clientId) =>
(Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
case ExplicitUserDefaultClientIdEntity(user) =>
(Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), Some(ClientQuotaManager.DefaultClientIdEntity))
case DefaultUserExplicitClientIdEntity(clientId) =>
(Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
case DefaultUserDefaultClientIdEntity =>
(Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.DefaultClientIdEntity))
case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
}
```
##########
core/src/main/scala/kafka/server/ClientQuotaManager.scala:
##########
@@ -55,7 +55,7 @@ object QuotaTypes {
object ClientQuotaManager {
// Purge sensors after 1 hour of inactivity
val InactiveSensorExpirationTimeSeconds = 3600
-
+ private val DefaultString = "<default>"
Review Comment:
How about renaming it to `DefaultName`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1939451332
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
I have a follow-up ask, while the two levels are still present they will no longer be under the path specified in this file since the path is ZK-specific. As long as we don't use this comment to generate documentation I am okay with this being removed in a subsequent PR. However, if we use this file to generate documentation could you change this as part of this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18641:
URL: https://github.com/apache/kafka/pull/18641#issuecomment-2624406797
@m1a2st could you please rebase code to include the recent flaky fixes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1937322084
##########
core/src/main/scala/kafka/server/ClientQuotaManager.scala:
##########
@@ -451,10 +449,31 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
}
}
+ private def getOrDefaultClient(
+ sanitizedClientId: Option[ClientQuotaEntity.ConfigEntity]
+ ): Option[ClientQuotaEntity.ConfigEntity] = {
+ if (sanitizedClientId.isEmpty)
+ None
+ else if (sanitizedClientId.get.name() == DefaultString)
+ Some(DefaultClientIdEntity)
+ else {
+ val clientId = sanitizedClientId.map(s => Sanitizer.desanitize(s.name()))
Review Comment:
we should handle the `ClientIdEntity` only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1944150081
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -145,27 +149,42 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, sanitizedClientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ sanitizedClientEntity = sanitizedClientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ quotaEntity match {
+ case UserEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
+ case DefaultUserEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), None)
+ case ClientIdEntity(clientId) =>
+ (None, Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
Review Comment:
After under discussion, We should remove this `sanitize`, it is unused in this path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1942971563
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -145,27 +149,42 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, sanitizedClientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ sanitizedClientEntity = sanitizedClientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ quotaEntity match {
+ case UserEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
+ case DefaultUserEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), None)
+ case ClientIdEntity(clientId) =>
+ (None, Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
Review Comment:
In sanitize method, It deal with `*` and `+` symbols. Therefore, I think it is not only handling matters related to Zookeeper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1942971563
##########
core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala:
##########
@@ -145,27 +149,42 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, sanitizedClientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ sanitizedClientEntity = sanitizedClientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ quotaEntity match {
+ case UserEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
+ case DefaultUserEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), None)
+ case ClientIdEntity(clientId) =>
+ (None, Some(ClientQuotaManager.ClientIdEntity(Sanitizer.sanitize(clientId))))
Review Comment:
In `sanitize` method, It deal with `*` and `+` symbols. Therefore, I think it is not only handling matters related to Zookeeper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1922560281
##########
server-common/src/main/java/org/apache/kafka/server/config/KraftInternals.java:
##########
@@ -16,12 +16,11 @@
*/
package org.apache.kafka.server.config;
-public class ZooKeeperInternals {
+public class KraftInternals {
Review Comment:
I don't think this name makes sense - there's nothing specific to kraft in this code. Can we move this constant to one of the quota classes and have the others reference it from there?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Rename ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1922770484
##########
server-common/src/main/java/org/apache/kafka/server/config/KraftInternals.java:
##########
@@ -16,12 +16,11 @@
*/
package org.apache.kafka.server.config;
-public class ZooKeeperInternals {
+public class KraftInternals {
Review Comment:
@m1a2st Could you please clean up the usage of `ZooKeeperInternals.DEFAULT_STRING` first? We are currently using `ZooKeeperInternals.DEFAULT_STRING` as a specific string for compatibility with the zk handler (QuotaConfigHandler). Since #18617 will remove `QuotaConfigHandler`, the kraft `ClientQuotaMetadataManager` no longer needs to use `ZooKeeperInternals.DEFAULT_STRING`.
After addressing above comment, we can move `<default>` to be a internal constant of `ClientQuotaManager` to ensure the compatibility of metrics `kafka.server:type={Produce|Fetch},user=([-.\w]+),client-id=([-.\w]+)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1940475710
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
I think those comments in DynamicBrokerConfig are not used to generate documentation, but it will be better to correct it - as least, the description about zk path must be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1941028115
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
Okay, then I will review once they have been changed 😊!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1941076696
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
Thanks for @clolov and @chia7712, update this document. If there is any idea, please let me know
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18530 Remove ZooKeeperInternals [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #18641:
URL: https://github.com/apache/kafka/pull/18641#discussion_r1939451332
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -48,13 +48,6 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
Review Comment:
Heya @m1a2st! I have a follow-up ask, while the two levels are still present they will no longer be under the path specified in this file since the path is ZK-specific. As long as we don't use this comment to generate documentation I am okay with this being removed in a subsequent PR. However, if we use this file to generate documentation could you change this as part of this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org