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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
