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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


Reply via email to