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 2024/06/18 06:45:33 UTC
[PR] ConfigCommand should not allow to define both `entity-default` and `entity-name` [kafka]
m1a2st opened a new pull request, #16381:
URL: https://github.com/apache/kafka/pull/16381
https://issues.apache.org/jira/browse/KAFKA-16959
add new check and test fot can't allow both parameter
waiting for https://github.com/apache/kafka/pull/16317
### 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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2374632265
Can you please correct the title as per the jira i.e. .....should `not` allow.....
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2375219362
> Can you please correct the title as per the jira i.e. .....should not allow.....
After discussion, we've decided to allow defining both entity-default and entity-name. This makes sense, as Admin#incrementalAlterConfigs supports it, so there's no reason to disable it in the command tool. @m1a2st Could you please correct the title for the jira?
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2210438103
@chia7712, Thanks for your comments, I will take a look at aother type.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16381:
URL: https://github.com/apache/kafka/pull/16381#discussion_r1665952364
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -396,9 +396,11 @@ object ConfigCommand extends Logging {
throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
val newConfig = new JConfig(newEntries.asJava.values)
- val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ entityNames.foreach { entityName =>
+ val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
Review Comment:
@m1a2st any update about this comment?
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2375431411
@chia7712, @apoorvmittal10, Thanks for your reminder, I will update the Jira Title.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-4028541548
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2210434480
@m1a2st Could you take a look at other ConfigTypes? I feel they can support both entity-default and entity-name too
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2212341885
@chia7712, Thanks for your comments, PTAL
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-5087358306
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16381:
URL: https://github.com/apache/kafka/pull/16381#discussion_r1673154305
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -354,128 +354,150 @@ object ConfigCommand extends Logging {
}
}
- @nowarn("cat=deprecation")
def alterConfig(adminClient: Admin, opts: ConfigCommandOptions): Unit = {
val entityTypes = opts.entityTypes
val entityNames = opts.entityNames
val entityTypeHead = entityTypes.head
- val entityNameHead = entityNames.head
val configsToBeAddedMap = parseConfigsToBeAdded(opts).asScala.toMap // no need for mutability
val configsToBeAdded = configsToBeAddedMap.map { case (k, v) => (k, new ConfigEntry(k, v)) }
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
entityTypeHead match {
case ConfigType.TOPIC =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
+ alterTopicConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.BROKER =>
+ alterBrokerConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case BrokerLoggerConfigType =>
+ alterBrokerLoggingConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.USER | ConfigType.CLIENT =>
+ alterUserOrClientConfig(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.IP =>
+ alterIpConfig(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.CLIENT_METRICS =>
+ alterClientMetricsConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
+ }
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ if (entityNames.nonEmpty) {
+ entityNames.foreach(
+ entityName => println(s"Completed updating config for ${entityTypeHead.dropRight(1)} $entityName.")
+ )
+ } else
+ println(s"Completed updating default config for $entityTypeHead in the cluster.")
+ }
- val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ private def alterTopicConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
+ }
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(entityNames.map(new ConfigResource(ConfigResource.Type.TOPIC, _))
+ .map(_ -> alterEntries).toMap.asJava, alterOptions)
+ .all().get(60, TimeUnit.SECONDS)
+ }
- case ConfigType.BROKER =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
+ private def getOldConfig(adminClient: Admin, entityTypeHead: String, configsToBeDeleted: Seq[String], entityName: String): Map[String, ConfigEntry] = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ oldConfig
+ }
- val newEntries = oldConfig ++ configsToBeAdded -- configsToBeDeleted
- val sensitiveEntries = newEntries.filter(_._2.value == null)
- if (sensitiveEntries.nonEmpty)
- throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
- val newConfig = new JConfig(newEntries.asJava.values)
+ @nowarn("cat=deprecation")
+ private def alterBrokerConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]): Unit = {
+ entityNames.foreach { entityName =>
+ val oldConfig = getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
- val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ val newEntries = oldConfig ++ configsToBeAdded -- configsToBeDeleted
+ val sensitiveEntries = newEntries.filter(_._2.value == null)
+ if (sensitiveEntries.nonEmpty)
+ throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
+ val newConfig = new JConfig(newEntries.asJava.values)
- case BrokerLoggerConfigType =>
- val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = true, describeAll = false).map(_.name)
- // fail the command if any of the configured broker loggers do not exist
- val invalidBrokerLoggers = configsToBeDeleted.filterNot(validLoggers.contains) ++ configsToBeAdded.keys.filterNot(validLoggers.contains)
- if (invalidBrokerLoggers.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid broker logger(s): ${invalidBrokerLoggers.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.BROKER_LOGGER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterLogLevelEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterLogLevelEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
+ }
- case ConfigType.USER | ConfigType.CLIENT =>
- val hasQuotaConfigsToAdd = configsToBeAdded.keys.exists(QuotaConfigs.isClientOrUserQuotaConfig)
- val scramConfigsToAddMap = configsToBeAdded.filter(entry => ScramMechanism.isScram(entry._1))
- val unknownConfigsToAdd = configsToBeAdded.keys.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
- val hasQuotaConfigsToDelete = configsToBeDeleted.exists(QuotaConfigs.isClientOrUserQuotaConfig)
- val scramConfigsToDelete = configsToBeDeleted.filter(ScramMechanism.isScram)
- val unknownConfigsToDelete = configsToBeDeleted.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
- if (entityTypeHead == ConfigType.CLIENT || entityTypes.size == 2) { // size==2 for case where users is specified first on the command line, before clients
- // either just a client or both a user and a client
- if (unknownConfigsToAdd.nonEmpty || scramConfigsToAddMap.nonEmpty)
- throw new IllegalArgumentException(s"Only quota configs can be added for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToAdd ++ scramConfigsToAddMap.keys}")
- if (unknownConfigsToDelete.nonEmpty || scramConfigsToDelete.nonEmpty)
- throw new IllegalArgumentException(s"Only quota configs can be deleted for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToDelete ++ scramConfigsToDelete}")
- } else { // ConfigType.User
- if (unknownConfigsToAdd.nonEmpty)
- throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be added for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToAdd")
- if (unknownConfigsToDelete.nonEmpty)
- throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be deleted for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToDelete")
- if (scramConfigsToAddMap.nonEmpty || scramConfigsToDelete.nonEmpty) {
- if (entityNames.exists(_.isEmpty)) // either --entity-type users --entity-default or --user-defaults
- throw new IllegalArgumentException("The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server.")
- if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete)
- throw new IllegalArgumentException(s"Cannot alter both quota and SCRAM credential configs simultaneously for '${ConfigType.USER}' using --bootstrap-server.")
- }
- }
+ private def alterBrokerLoggingConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = true, describeAll = false).map(_.name)
+ // fail the command if any of the configured broker loggers do not exist
+ val invalidBrokerLoggers = configsToBeDeleted.filterNot(validLoggers.contains) ++ configsToBeAdded.keys.filterNot(validLoggers.contains)
+ if (invalidBrokerLoggers.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid broker logger(s): ${invalidBrokerLoggers.mkString(",")}")
+ }
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterLogLevelEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(entityNames.map(new ConfigResource(ConfigResource.Type.BROKER_LOGGER, _))
+ .map(_ -> alterLogLevelEntries).toMap.asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
- if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete) {
- alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
- } else {
- // handle altering user SCRAM credential configs
- if (entityNames.size != 1)
- // should never happen, if we get here then it is a bug
- throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames")
- alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete)
- }
- case ConfigType.IP =>
- val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key))
- if (unknownConfigs.nonEmpty)
- throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.IP}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
- alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
- case ConfigType.CLIENT_METRICS =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
- case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
+ private def alterUserOrClientConfig(adminClient: Admin, entityTypes: List[String], entityNames: List[String], configsToBeAddedMap: Predef.Map[String, String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ val entityTypeHead = entityTypes.head
+ val hasQuotaConfigsToAdd = configsToBeAdded.keys.exists(QuotaConfigs.isClientOrUserQuotaConfig)
+ val scramConfigsToAddMap = configsToBeAdded.filter(entry => ScramMechanism.isScram(entry._1))
+ val unknownConfigsToAdd = configsToBeAdded.keys.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
+ val hasQuotaConfigsToDelete = configsToBeDeleted.exists(QuotaConfigs.isClientOrUserQuotaConfig)
+ val scramConfigsToDelete = configsToBeDeleted.filter(ScramMechanism.isScram)
+ val unknownConfigsToDelete = configsToBeDeleted.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
+ if (entityTypeHead == ConfigType.CLIENT || entityTypes.size == 2) { // size==2 for case where users is specified first on the command line, before clients
+ // either just a client or both a user and a client
+ if (unknownConfigsToAdd.nonEmpty || scramConfigsToAddMap.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota configs can be added for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToAdd ++ scramConfigsToAddMap.keys}")
+ if (unknownConfigsToDelete.nonEmpty || scramConfigsToDelete.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota configs can be deleted for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToDelete ++ scramConfigsToDelete}")
+ } else { // ConfigType.User
+ if (unknownConfigsToAdd.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be added for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToAdd")
+ if (unknownConfigsToDelete.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be deleted for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToDelete")
+ if (scramConfigsToAddMap.nonEmpty || scramConfigsToDelete.nonEmpty) {
+ if (entityNames.exists(_.isEmpty)) // either --entity-type users --entity-default or --user-defaults
+ throw new IllegalArgumentException("The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server.")
+ if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete)
+ throw new IllegalArgumentException(s"Cannot alter both quota and SCRAM credential configs simultaneously for '${ConfigType.USER}' using --bootstrap-server.")
+ }
}
- if (entityNameHead.nonEmpty)
- println(s"Completed updating config for ${entityTypeHead.dropRight(1)} $entityNameHead.")
- else
- println(s"Completed updating default config for $entityTypeHead in the cluster.")
+ if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete) {
+ alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
+ } else {
+ // handle altering user SCRAM credential configs
+ if (entityNames.size != 1)
+ // should never happen, if we get here then it is a bug
+ throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames")
+ alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete)
+ }
+ }
+
+ private def alterIpConfig(adminClient: Admin, entityTypes: List[String], entityNames: List[String], configsToBeAddedMap: Predef.Map[String, String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key))
+ if (unknownConfigs.nonEmpty)
+ throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.IP}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
+ alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
+ }
+
+ private def alterClientMetricsConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
Review Comment:
BTW, could you add test for `InvalidConfigurationException`?
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -354,128 +354,150 @@ object ConfigCommand extends Logging {
}
}
- @nowarn("cat=deprecation")
def alterConfig(adminClient: Admin, opts: ConfigCommandOptions): Unit = {
val entityTypes = opts.entityTypes
val entityNames = opts.entityNames
val entityTypeHead = entityTypes.head
- val entityNameHead = entityNames.head
val configsToBeAddedMap = parseConfigsToBeAdded(opts).asScala.toMap // no need for mutability
val configsToBeAdded = configsToBeAddedMap.map { case (k, v) => (k, new ConfigEntry(k, v)) }
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
entityTypeHead match {
case ConfigType.TOPIC =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
+ alterTopicConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.BROKER =>
+ alterBrokerConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case BrokerLoggerConfigType =>
+ alterBrokerLoggingConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.USER | ConfigType.CLIENT =>
+ alterUserOrClientConfig(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.IP =>
+ alterIpConfig(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeAdded, configsToBeDeleted)
+ case ConfigType.CLIENT_METRICS =>
+ alterClientMetricsConfig(adminClient, entityTypeHead, entityNames, configsToBeAdded, configsToBeDeleted)
+ case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
+ }
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ if (entityNames.nonEmpty) {
+ entityNames.foreach(
+ entityName => println(s"Completed updating config for ${entityTypeHead.dropRight(1)} $entityName.")
+ )
+ } else
+ println(s"Completed updating default config for $entityTypeHead in the cluster.")
+ }
- val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ private def alterTopicConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
+ }
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(entityNames.map(new ConfigResource(ConfigResource.Type.TOPIC, _))
+ .map(_ -> alterEntries).toMap.asJava, alterOptions)
+ .all().get(60, TimeUnit.SECONDS)
+ }
- case ConfigType.BROKER =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
+ private def getOldConfig(adminClient: Admin, entityTypeHead: String, configsToBeDeleted: Seq[String], entityName: String): Map[String, ConfigEntry] = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ oldConfig
+ }
- val newEntries = oldConfig ++ configsToBeAdded -- configsToBeDeleted
- val sensitiveEntries = newEntries.filter(_._2.value == null)
- if (sensitiveEntries.nonEmpty)
- throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
- val newConfig = new JConfig(newEntries.asJava.values)
+ @nowarn("cat=deprecation")
+ private def alterBrokerConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]): Unit = {
+ entityNames.foreach { entityName =>
+ val oldConfig = getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
- val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ val newEntries = oldConfig ++ configsToBeAdded -- configsToBeDeleted
+ val sensitiveEntries = newEntries.filter(_._2.value == null)
+ if (sensitiveEntries.nonEmpty)
+ throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
+ val newConfig = new JConfig(newEntries.asJava.values)
- case BrokerLoggerConfigType =>
- val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = true, describeAll = false).map(_.name)
- // fail the command if any of the configured broker loggers do not exist
- val invalidBrokerLoggers = configsToBeDeleted.filterNot(validLoggers.contains) ++ configsToBeAdded.keys.filterNot(validLoggers.contains)
- if (invalidBrokerLoggers.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid broker logger(s): ${invalidBrokerLoggers.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.BROKER_LOGGER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterLogLevelEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterLogLevelEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
+ }
- case ConfigType.USER | ConfigType.CLIENT =>
- val hasQuotaConfigsToAdd = configsToBeAdded.keys.exists(QuotaConfigs.isClientOrUserQuotaConfig)
- val scramConfigsToAddMap = configsToBeAdded.filter(entry => ScramMechanism.isScram(entry._1))
- val unknownConfigsToAdd = configsToBeAdded.keys.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
- val hasQuotaConfigsToDelete = configsToBeDeleted.exists(QuotaConfigs.isClientOrUserQuotaConfig)
- val scramConfigsToDelete = configsToBeDeleted.filter(ScramMechanism.isScram)
- val unknownConfigsToDelete = configsToBeDeleted.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
- if (entityTypeHead == ConfigType.CLIENT || entityTypes.size == 2) { // size==2 for case where users is specified first on the command line, before clients
- // either just a client or both a user and a client
- if (unknownConfigsToAdd.nonEmpty || scramConfigsToAddMap.nonEmpty)
- throw new IllegalArgumentException(s"Only quota configs can be added for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToAdd ++ scramConfigsToAddMap.keys}")
- if (unknownConfigsToDelete.nonEmpty || scramConfigsToDelete.nonEmpty)
- throw new IllegalArgumentException(s"Only quota configs can be deleted for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToDelete ++ scramConfigsToDelete}")
- } else { // ConfigType.User
- if (unknownConfigsToAdd.nonEmpty)
- throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be added for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToAdd")
- if (unknownConfigsToDelete.nonEmpty)
- throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be deleted for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToDelete")
- if (scramConfigsToAddMap.nonEmpty || scramConfigsToDelete.nonEmpty) {
- if (entityNames.exists(_.isEmpty)) // either --entity-type users --entity-default or --user-defaults
- throw new IllegalArgumentException("The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server.")
- if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete)
- throw new IllegalArgumentException(s"Cannot alter both quota and SCRAM credential configs simultaneously for '${ConfigType.USER}' using --bootstrap-server.")
- }
- }
+ private def alterBrokerLoggingConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = true, describeAll = false).map(_.name)
+ // fail the command if any of the configured broker loggers do not exist
+ val invalidBrokerLoggers = configsToBeDeleted.filterNot(validLoggers.contains) ++ configsToBeAdded.keys.filterNot(validLoggers.contains)
+ if (invalidBrokerLoggers.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid broker logger(s): ${invalidBrokerLoggers.mkString(",")}")
+ }
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterLogLevelEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(entityNames.map(new ConfigResource(ConfigResource.Type.BROKER_LOGGER, _))
+ .map(_ -> alterLogLevelEntries).toMap.asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
- if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete) {
- alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
- } else {
- // handle altering user SCRAM credential configs
- if (entityNames.size != 1)
- // should never happen, if we get here then it is a bug
- throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames")
- alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete)
- }
- case ConfigType.IP =>
- val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key))
- if (unknownConfigs.nonEmpty)
- throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.IP}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
- alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
- case ConfigType.CLIENT_METRICS =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
- case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
+ private def alterUserOrClientConfig(adminClient: Admin, entityTypes: List[String], entityNames: List[String], configsToBeAddedMap: Predef.Map[String, String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ val entityTypeHead = entityTypes.head
+ val hasQuotaConfigsToAdd = configsToBeAdded.keys.exists(QuotaConfigs.isClientOrUserQuotaConfig)
+ val scramConfigsToAddMap = configsToBeAdded.filter(entry => ScramMechanism.isScram(entry._1))
+ val unknownConfigsToAdd = configsToBeAdded.keys.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
+ val hasQuotaConfigsToDelete = configsToBeDeleted.exists(QuotaConfigs.isClientOrUserQuotaConfig)
+ val scramConfigsToDelete = configsToBeDeleted.filter(ScramMechanism.isScram)
+ val unknownConfigsToDelete = configsToBeDeleted.filterNot(key => ScramMechanism.isScram(key) || QuotaConfigs.isClientOrUserQuotaConfig(key))
+ if (entityTypeHead == ConfigType.CLIENT || entityTypes.size == 2) { // size==2 for case where users is specified first on the command line, before clients
+ // either just a client or both a user and a client
+ if (unknownConfigsToAdd.nonEmpty || scramConfigsToAddMap.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota configs can be added for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToAdd ++ scramConfigsToAddMap.keys}")
+ if (unknownConfigsToDelete.nonEmpty || scramConfigsToDelete.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota configs can be deleted for '${ConfigType.CLIENT}' using --bootstrap-server. Unexpected config names: ${unknownConfigsToDelete ++ scramConfigsToDelete}")
+ } else { // ConfigType.User
+ if (unknownConfigsToAdd.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be added for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToAdd")
+ if (unknownConfigsToDelete.nonEmpty)
+ throw new IllegalArgumentException(s"Only quota and SCRAM credential configs can be deleted for '${ConfigType.USER}' using --bootstrap-server. Unexpected config names: $unknownConfigsToDelete")
+ if (scramConfigsToAddMap.nonEmpty || scramConfigsToDelete.nonEmpty) {
+ if (entityNames.exists(_.isEmpty)) // either --entity-type users --entity-default or --user-defaults
+ throw new IllegalArgumentException("The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server.")
+ if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete)
+ throw new IllegalArgumentException(s"Cannot alter both quota and SCRAM credential configs simultaneously for '${ConfigType.USER}' using --bootstrap-server.")
+ }
}
- if (entityNameHead.nonEmpty)
- println(s"Completed updating config for ${entityTypeHead.dropRight(1)} $entityNameHead.")
- else
- println(s"Completed updating default config for $entityTypeHead in the cluster.")
+ if (hasQuotaConfigsToAdd || hasQuotaConfigsToDelete) {
+ alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
+ } else {
+ // handle altering user SCRAM credential configs
+ if (entityNames.size != 1)
+ // should never happen, if we get here then it is a bug
+ throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames")
+ alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete)
+ }
+ }
+
+ private def alterIpConfig(adminClient: Admin, entityTypes: List[String], entityNames: List[String], configsToBeAddedMap: Predef.Map[String, String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key))
+ if (unknownConfigs.nonEmpty)
+ throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.IP}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
+ alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
+ }
+
+ private def alterClientMetricsConfig(adminClient: Admin, entityTypeHead: String, entityNames: List[String], configsToBeAdded: Predef.Map[String, ConfigEntry], configsToBeDeleted: Seq[String]) = {
+ entityNames.foreach { entityName =>
+ getOldConfig(adminClient, entityTypeHead, configsToBeDeleted, entityName)
Review Comment:
I guess you want to reuse the validation, right? However, the method name "get" is a bit weird since it is used in the "foreach". Could you please do a bit refactor?
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2222873610
@chia7712, add new test for `InvalidConfigurationException`, PTAL, Thanks
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-5434710765
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2368834415
rebase on the trunk, and fix the conflict
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16381:
URL: https://github.com/apache/kafka/pull/16381#issuecomment-2185156040
Gentle ping, @chia7712, I update the title
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #16381:
URL: https://github.com/apache/kafka/pull/16381#discussion_r1666501123
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -396,9 +396,11 @@ object ConfigCommand extends Logging {
throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
val newConfig = new JConfig(newEntries.asJava.values)
- val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ entityNames.foreach { entityName =>
+ val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
Review Comment:
No, I think I doesn't find the key point in `ConfigType.BROKER` this case.
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16381:
URL: https://github.com/apache/kafka/pull/16381#discussion_r1668602936
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -359,61 +359,64 @@ object ConfigCommand extends Logging {
val entityTypes = opts.entityTypes
val entityNames = opts.entityNames
val entityTypeHead = entityTypes.head
- val entityNameHead = entityNames.head
val configsToBeAddedMap = parseConfigsToBeAdded(opts).asScala.toMap // no need for mutability
val configsToBeAdded = configsToBeAddedMap.map { case (k, v) => (k, new ConfigEntry(k, v)) }
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
entityTypeHead match {
case ConfigType.TOPIC =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
-
+ entityNames.foreach { entityName =>
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
+
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+
+ val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
case ConfigType.BROKER =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val newEntries = oldConfig ++ configsToBeAdded -- configsToBeDeleted
- val sensitiveEntries = newEntries.filter(_._2.value == null)
- if (sensitiveEntries.nonEmpty)
- throw new InvalidConfigurationException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
- val newConfig = new JConfig(newEntries.asJava.values)
-
- val configResource = new ConfigResource(ConfigResource.Type.BROKER, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ entityNames.foreach { entityName =>
Review Comment:
Could you please move those code to separate method? That can reduce the complexity and it can be test easily.
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -359,61 +359,64 @@ object ConfigCommand extends Logging {
val entityTypes = opts.entityTypes
val entityNames = opts.entityNames
val entityTypeHead = entityTypes.head
- val entityNameHead = entityNames.head
val configsToBeAddedMap = parseConfigsToBeAdded(opts).asScala.toMap // no need for mutability
val configsToBeAdded = configsToBeAddedMap.map { case (k, v) => (k, new ConfigEntry(k, v)) }
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
entityTypeHead match {
case ConfigType.TOPIC =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
-
+ entityNames.foreach { entityName =>
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityName, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
+
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+
+ val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityName)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
Review Comment:
Could you please leverage `incrementalAlterConfigs` to send the requests together instead of loop?
--
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-16959: ConfigCommand should allow to define both `entity-default` and `entity-name` [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #16381: KAFKA-16959: ConfigCommand should allow to define both `entity-default` and `entity-name`
URL: https://github.com/apache/kafka/pull/16381
--
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