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