You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/08/29 00:39:37 UTC
[GitHub] [kafka] dielhennr opened a new pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols [REWRITE]
dielhennr opened a new pull request #11281:
URL: https://github.com/apache/kafka/pull/11281
A rewrite of KAFKA-12886.
This PR does:
1. Enables request forwarding to the controller by default for zookeeper mutation protocols.
2. Since only the broker that a dynamic config applies to can validate its config update, this pr validates the configs from {Alter, IncrementalAlter}ConfigsRequest before forwarding them to both Zk and KRaft controllers.
3. Refactor tests that did not expect zookeeper brokers to forward requests.
--
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
[GitHub] [kafka] dielhennr commented on a change in pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols in IBP 3.1 and higher [REWRITE]
Posted by GitBox <gi...@apache.org>.
dielhennr commented on a change in pull request #11281:
URL: https://github.com/apache/kafka/pull/11281#discussion_r697964085
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -2622,6 +2623,32 @@ class KafkaApis(val requestChannel: RequestChannel,
.setTopics(endOffsetsForAllTopics)))
}
+ def handleDynamicConfigs(request: RequestChannel.Request, isIncrementalRequest: Boolean): Unit = {
Review comment:
This handles the validation, persisting, and forwarding logic for both ZK and KRaft brokers to ensure that dynamic broker configs are validated on the broker that the configs are for before forwarding them to the controller to be persisted.
--
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
[GitHub] [kafka] dielhennr commented on a change in pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols [REWRITE]
Posted by GitBox <gi...@apache.org>.
dielhennr commented on a change in pull request #11281:
URL: https://github.com/apache/kafka/pull/11281#discussion_r697942506
##########
File path: core/src/main/scala/kafka/server/ConfigHelper.scala
##########
@@ -22,20 +22,127 @@ import java.util.{Collections, Properties}
import kafka.log.LogConfig
import kafka.server.metadata.ConfigRepository
import kafka.utils.{Log4jController, Logging}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidRequestException}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.message.DescribeConfigsResponseData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ApiError, DescribeConfigsResponse}
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource
-import scala.collection.{Map, mutable}
+import scala.collection.{Map, mutable, Seq}
import scala.jdk.CollectionConverters._
class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
+ def getBrokerId(resource: ConfigResource) = {
Review comment:
`getBrokerId`, `prepareIncrementalConfigs`, `validateLogLevelConfigs`, and `toLoggableProps` were all helper methods in `ZkAdminManager`. I've moved them to this class (`ConfigHelper`) to help with validating broker configs before they are forwarded to be persisted.
##########
File path: core/src/main/scala/kafka/server/ConfigHelper.scala
##########
@@ -22,20 +22,127 @@ import java.util.{Collections, Properties}
import kafka.log.LogConfig
import kafka.server.metadata.ConfigRepository
import kafka.utils.{Log4jController, Logging}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidRequestException}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.message.DescribeConfigsResponseData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ApiError, DescribeConfigsResponse}
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource
-import scala.collection.{Map, mutable}
+import scala.collection.{Map, mutable, Seq}
import scala.jdk.CollectionConverters._
class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
+ def getBrokerId(resource: ConfigResource) = {
+ if (resource.name == null || resource.name.isEmpty)
+ None
+ else {
+ Some(resourceNameToBrokerId(resource.name))
+ }
+ }
+
+ def getAndValidateBrokerId(resource: ConfigResource) = {
+ val id = getBrokerId(resource)
+ if (id.nonEmpty && (id.get != this.config.brokerId))
+ throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received ${resource.name}")
+ id
+ }
+
+ def validateBrokerConfigs(resource: ConfigResource,
Review comment:
This method does the validation part of `alterBrokerConfig` from `ZkAdminManager` without persisting the configs in Zookeeper. This allows validation of dynamic configs on brokers before they are forwarded for both ZK and KRaft clusters.
The validation part of `alterBrokerConfig` is removed in this PR and `alterBrokerConfig` now only persists the configs. This is done because the request will only end up at the controller if the validation on the proxy broker succeeded, and all that is left to do at that point is to persist the configs.
##########
File path: core/src/main/scala/kafka/server/ZkAdminManager.scala
##########
@@ -531,24 +518,21 @@ class ZkAdminManager(val config: KafkaConfig,
resource.`type` match {
case ConfigResource.Type.TOPIC =>
val configProps = adminZkClient.fetchEntityConfig(ConfigType.Topic, resource.name)
- prepareIncrementalConfigs(alterConfigOps, configProps, LogConfig.configKeys)
+ configHelper.prepareIncrementalConfigs(alterConfigOps, configProps, LogConfig.configKeys)
alterTopicConfigs(resource, validateOnly, configProps, configEntriesMap)
case ConfigResource.Type.BROKER =>
- val brokerId = getBrokerId(resource)
+ val brokerId = configHelper.getBrokerId(resource)
val perBrokerConfig = brokerId.nonEmpty
val persistentProps = if (perBrokerConfig) adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.get.toString)
else adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default)
val configProps = this.config.dynamicConfig.fromPersistentProps(persistentProps, perBrokerConfig)
- prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
+ configHelper.prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
alterBrokerConfigs(resource, validateOnly, configProps, configEntriesMap)
case ConfigResource.Type.BROKER_LOGGER =>
- getBrokerId(resource)
- validateLogLevelConfigs(alterConfigOps)
Review comment:
The Zk Controller node will be the only one calling the method `incrementalAlterConfigs` since this is only called if the request was forwarded. This validation step is being removed since it is being done on the brokers.
--
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
[GitHub] [kafka] dielhennr commented on a change in pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols in IBP 3.1 and higher [REWRITE]
Posted by GitBox <gi...@apache.org>.
dielhennr commented on a change in pull request #11281:
URL: https://github.com/apache/kafka/pull/11281#discussion_r697942051
##########
File path: core/src/main/scala/kafka/server/ZkAdminManager.scala
##########
@@ -531,24 +518,21 @@ class ZkAdminManager(val config: KafkaConfig,
resource.`type` match {
case ConfigResource.Type.TOPIC =>
val configProps = adminZkClient.fetchEntityConfig(ConfigType.Topic, resource.name)
- prepareIncrementalConfigs(alterConfigOps, configProps, LogConfig.configKeys)
+ configHelper.prepareIncrementalConfigs(alterConfigOps, configProps, LogConfig.configKeys)
alterTopicConfigs(resource, validateOnly, configProps, configEntriesMap)
case ConfigResource.Type.BROKER =>
- val brokerId = getBrokerId(resource)
+ val brokerId = configHelper.getBrokerId(resource)
val perBrokerConfig = brokerId.nonEmpty
val persistentProps = if (perBrokerConfig) adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.get.toString)
else adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default)
val configProps = this.config.dynamicConfig.fromPersistentProps(persistentProps, perBrokerConfig)
- prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
+ configHelper.prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
alterBrokerConfigs(resource, validateOnly, configProps, configEntriesMap)
case ConfigResource.Type.BROKER_LOGGER =>
- getBrokerId(resource)
- validateLogLevelConfigs(alterConfigOps)
Review comment:
The Zk Controller node will be the only one calling the method `incrementalAlterConfigs` base on the new logic in `handleDynamicConfigs.` This validation step is being removed since it is being done on the brokers.
--
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
[GitHub] [kafka] dielhennr commented on a change in pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols in IBP 3.1 and higher [REWRITE]
Posted by GitBox <gi...@apache.org>.
dielhennr commented on a change in pull request #11281:
URL: https://github.com/apache/kafka/pull/11281#discussion_r697942602
##########
File path: core/src/main/scala/kafka/server/ConfigHelper.scala
##########
@@ -22,20 +22,127 @@ import java.util.{Collections, Properties}
import kafka.log.LogConfig
import kafka.server.metadata.ConfigRepository
import kafka.utils.{Log4jController, Logging}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidRequestException}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.message.DescribeConfigsResponseData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ApiError, DescribeConfigsResponse}
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource
-import scala.collection.{Map, mutable}
+import scala.collection.{Map, mutable, Seq}
import scala.jdk.CollectionConverters._
class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
+ def getBrokerId(resource: ConfigResource) = {
+ if (resource.name == null || resource.name.isEmpty)
+ None
+ else {
+ Some(resourceNameToBrokerId(resource.name))
+ }
+ }
+
+ def getAndValidateBrokerId(resource: ConfigResource) = {
+ val id = getBrokerId(resource)
+ if (id.nonEmpty && (id.get != this.config.brokerId))
+ throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received ${resource.name}")
+ id
+ }
+
+ def validateBrokerConfigs(resource: ConfigResource,
Review comment:
This method does the validation part of `alterBrokerConfig` from `ZkAdminManager` without persisting the configs in Zookeeper. This allows validation of dynamic configs on brokers before they are forwarded to the controller for both ZK and KRaft clusters.
The validation part of `alterBrokerConfig` is removed in this PR and `alterBrokerConfig` now only persists the configs. This is done because the request will only end up at the controller if the validation on the proxy broker succeeded, and all that is left to do at that point is to persist the configs.
--
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
[GitHub] [kafka] dielhennr commented on a change in pull request #11281: KAFKA-12886: Enable request forwarding to the controller by default for zookeeper mutation protocols in IBP 3.1 and higher [REWRITE]
Posted by GitBox <gi...@apache.org>.
dielhennr commented on a change in pull request #11281:
URL: https://github.com/apache/kafka/pull/11281#discussion_r697964085
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -2622,6 +2623,32 @@ class KafkaApis(val requestChannel: RequestChannel,
.setTopics(endOffsetsForAllTopics)))
}
+ def handleDynamicConfigs(request: RequestChannel.Request, isIncrementalRequest: Boolean): Unit = {
Review comment:
This handles the validation, persisting, and forwarding logic for both ZK and KRaft brokers to ensure that configs are validated on the broker that the configs are for before persisting them.
##########
File path: core/src/main/scala/kafka/server/ConfigHelper.scala
##########
@@ -22,20 +22,145 @@ import java.util.{Collections, Properties}
import kafka.log.LogConfig
import kafka.server.metadata.ConfigRepository
import kafka.utils.{Log4jController, Logging}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidRequestException}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.message.DescribeConfigsResponseData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ApiError, DescribeConfigsResponse}
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource
-import scala.collection.{Map, mutable}
+import scala.collection.{Map, mutable, Seq}
import scala.jdk.CollectionConverters._
class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
+ def getBrokerId(resource: ConfigResource) = {
+ if (resource.name == null || resource.name.isEmpty)
+ None
+ else {
+ Some(resourceNameToBrokerId(resource.name))
+ }
+ }
+
+ def getAndValidateBrokerId(resource: ConfigResource) = {
+ val id = getBrokerId(resource)
+ if (id.nonEmpty && (id.get != this.config.brokerId))
+ throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received ${resource.name}")
+ id
+ }
+
+ def prepareIncrementalConfigs(alterConfigOps: Seq[AlterConfigOp], configProps: Properties, configKeys: Map[String, ConfigKey]): Unit = {
+
+ def listType(configName: String, configKeys: Map[String, ConfigKey]): Boolean = {
+ val configKey = configKeys(configName)
+ if (configKey == null)
+ throw new InvalidConfigurationException(s"Unknown topic config name: $configName")
+ configKey.`type` == ConfigDef.Type.LIST
+ }
+
+ alterConfigOps.foreach { alterConfigOp =>
+ val configPropName = alterConfigOp.configEntry.name
+ alterConfigOp.opType() match {
+ case OpType.SET => configProps.setProperty(alterConfigOp.configEntry.name, alterConfigOp.configEntry.value)
+ case OpType.DELETE => configProps.remove(alterConfigOp.configEntry.name)
+ case OpType.APPEND => {
+ if (!listType(alterConfigOp.configEntry.name, configKeys))
+ throw new InvalidRequestException(s"Config value append is not allowed for config key: ${alterConfigOp.configEntry.name}")
+ val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
+ .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
+ .getOrElse("")
+ .split(",").toList
+ val newValueList = oldValueList ::: alterConfigOp.configEntry.value.split(",").toList
+ configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
+ }
+ case OpType.SUBTRACT => {
+ if (!listType(alterConfigOp.configEntry.name, configKeys))
+ throw new InvalidRequestException(s"Config value subtract is not allowed for config key: ${alterConfigOp.configEntry.name}")
+ val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
+ .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
+ .getOrElse("")
+ .split(",").toList
+ val newValueList = oldValueList.diff(alterConfigOp.configEntry.value.split(",").toList)
+ configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
+ }
+ }
+ }
+ }
+
+ def validateLogLevelConfigs(alterConfigOps: Seq[AlterConfigOp]): Unit = {
+ def validateLoggerNameExists(loggerName: String): Unit = {
+ if (!Log4jController.loggerExists(loggerName))
+ throw new ConfigException(s"Logger $loggerName does not exist!")
+ }
+
+ alterConfigOps.foreach { alterConfigOp =>
+ val loggerName = alterConfigOp.configEntry.name
+ alterConfigOp.opType() match {
+ case OpType.SET =>
+ validateLoggerNameExists(loggerName)
+ val logLevel = alterConfigOp.configEntry.value
+ if (!LogLevelConfig.VALID_LOG_LEVELS.contains(logLevel)) {
+ val validLevelsStr = LogLevelConfig.VALID_LOG_LEVELS.asScala.mkString(", ")
+ throw new ConfigException(
+ s"Cannot set the log level of $loggerName to $logLevel as it is not a supported log level. " +
+ s"Valid log levels are $validLevelsStr"
+ )
+ }
+ case OpType.DELETE =>
+ validateLoggerNameExists(loggerName)
+ if (loggerName == Log4jController.ROOT_LOGGER)
+ throw new InvalidRequestException(s"Removing the log level of the ${Log4jController.ROOT_LOGGER} logger is not allowed")
+ case OpType.APPEND => throw new InvalidRequestException(s"${OpType.APPEND} operation is not allowed for the ${ConfigResource.Type.BROKER_LOGGER} resource")
+ case OpType.SUBTRACT => throw new InvalidRequestException(s"${OpType.SUBTRACT} operation is not allowed for the ${ConfigResource.Type.BROKER_LOGGER} resource")
+ }
+ }
+ }
+
+ def toLoggableProps(resource: ConfigResource, configProps: Properties): Map[String, String] = {
+ configProps.asScala.map {
+ case (key, value) => (key, KafkaConfig.loggableValue(resource.`type`, key, value))
+ }
+ }
+
+ def validateBrokerConfigs(resource: ConfigResource,
Review comment:
This method does the validation part of alterBrokerConfig from ZkAdminManager without persisting the configs in Zookeeper. This allows validation of dynamic configs on brokers before they are forwarded to the controller for both ZK and KRaft clusters.
The validation part of alterBrokerConfig is removed in this PR and alterBrokerConfig now only persists the configs. This is done because the request will only end up at the controller if the validation on the proxy broker succeeded, and all that is left to do at that point is to persist the configs.
##########
File path: core/src/main/scala/kafka/server/ConfigHelper.scala
##########
@@ -22,20 +22,145 @@ import java.util.{Collections, Properties}
import kafka.log.LogConfig
import kafka.server.metadata.ConfigRepository
import kafka.utils.{Log4jController, Logging}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidRequestException}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DescribeConfigsRequestData.DescribeConfigsResource
import org.apache.kafka.common.message.DescribeConfigsResponseData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ApiError, DescribeConfigsResponse}
import org.apache.kafka.common.requests.DescribeConfigsResponse.ConfigSource
-import scala.collection.{Map, mutable}
+import scala.collection.{Map, mutable, Seq}
import scala.jdk.CollectionConverters._
class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
+ def getBrokerId(resource: ConfigResource) = {
+ if (resource.name == null || resource.name.isEmpty)
+ None
+ else {
+ Some(resourceNameToBrokerId(resource.name))
+ }
+ }
+
+ def getAndValidateBrokerId(resource: ConfigResource) = {
+ val id = getBrokerId(resource)
+ if (id.nonEmpty && (id.get != this.config.brokerId))
+ throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received ${resource.name}")
+ id
+ }
+
+ def prepareIncrementalConfigs(alterConfigOps: Seq[AlterConfigOp], configProps: Properties, configKeys: Map[String, ConfigKey]): Unit = {
+
+ def listType(configName: String, configKeys: Map[String, ConfigKey]): Boolean = {
+ val configKey = configKeys(configName)
+ if (configKey == null)
+ throw new InvalidConfigurationException(s"Unknown topic config name: $configName")
+ configKey.`type` == ConfigDef.Type.LIST
+ }
+
+ alterConfigOps.foreach { alterConfigOp =>
+ val configPropName = alterConfigOp.configEntry.name
+ alterConfigOp.opType() match {
+ case OpType.SET => configProps.setProperty(alterConfigOp.configEntry.name, alterConfigOp.configEntry.value)
+ case OpType.DELETE => configProps.remove(alterConfigOp.configEntry.name)
+ case OpType.APPEND => {
+ if (!listType(alterConfigOp.configEntry.name, configKeys))
+ throw new InvalidRequestException(s"Config value append is not allowed for config key: ${alterConfigOp.configEntry.name}")
+ val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
+ .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
+ .getOrElse("")
+ .split(",").toList
+ val newValueList = oldValueList ::: alterConfigOp.configEntry.value.split(",").toList
+ configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
+ }
+ case OpType.SUBTRACT => {
+ if (!listType(alterConfigOp.configEntry.name, configKeys))
+ throw new InvalidRequestException(s"Config value subtract is not allowed for config key: ${alterConfigOp.configEntry.name}")
+ val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
+ .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
+ .getOrElse("")
+ .split(",").toList
+ val newValueList = oldValueList.diff(alterConfigOp.configEntry.value.split(",").toList)
+ configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
+ }
+ }
+ }
+ }
+
+ def validateLogLevelConfigs(alterConfigOps: Seq[AlterConfigOp]): Unit = {
+ def validateLoggerNameExists(loggerName: String): Unit = {
+ if (!Log4jController.loggerExists(loggerName))
+ throw new ConfigException(s"Logger $loggerName does not exist!")
+ }
+
+ alterConfigOps.foreach { alterConfigOp =>
+ val loggerName = alterConfigOp.configEntry.name
+ alterConfigOp.opType() match {
+ case OpType.SET =>
+ validateLoggerNameExists(loggerName)
+ val logLevel = alterConfigOp.configEntry.value
+ if (!LogLevelConfig.VALID_LOG_LEVELS.contains(logLevel)) {
+ val validLevelsStr = LogLevelConfig.VALID_LOG_LEVELS.asScala.mkString(", ")
+ throw new ConfigException(
+ s"Cannot set the log level of $loggerName to $logLevel as it is not a supported log level. " +
+ s"Valid log levels are $validLevelsStr"
+ )
+ }
+ case OpType.DELETE =>
+ validateLoggerNameExists(loggerName)
+ if (loggerName == Log4jController.ROOT_LOGGER)
+ throw new InvalidRequestException(s"Removing the log level of the ${Log4jController.ROOT_LOGGER} logger is not allowed")
+ case OpType.APPEND => throw new InvalidRequestException(s"${OpType.APPEND} operation is not allowed for the ${ConfigResource.Type.BROKER_LOGGER} resource")
+ case OpType.SUBTRACT => throw new InvalidRequestException(s"${OpType.SUBTRACT} operation is not allowed for the ${ConfigResource.Type.BROKER_LOGGER} resource")
+ }
+ }
+ }
+
+ def toLoggableProps(resource: ConfigResource, configProps: Properties): Map[String, String] = {
+ configProps.asScala.map {
+ case (key, value) => (key, KafkaConfig.loggableValue(resource.`type`, key, value))
+ }
+ }
+
+ def validateBrokerConfigs(resource: ConfigResource,
+ validateOnly: Boolean,
+ configProps: Properties): (ConfigResource, ApiError) = {
+ val brokerId = getAndValidateBrokerId(resource)
+ val perBrokerConfig = brokerId.nonEmpty
+ // Validate and process the reconfiguration
+ this.config.dynamicConfig.validate(configProps, perBrokerConfig)
+ if (!validateOnly) {
+ if (perBrokerConfig)
+ this.config.dynamicConfig.reloadUpdatedFilesWithoutConfigChange(configProps)
+ }
+
+ resource -> ApiError.NONE
+ }
+
+ def handleValidateBrokerConfigsError(exception: Throwable, resource: ConfigResource, configProps: Option[Properties], alterOps: Option[Seq[AlterConfigOp]]): (ConfigResource, ApiError) = {
Review comment:
Refactor error handling for dynamic config validation to be functional.
--
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