chia7712 commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1641801668
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -472,6 +472,131 @@ class DynamicConfigChangeTest extends
KafkaServerTestHarness {
}
}
+ private def setBrokerConfigs(brokerId: String, newValue: Long): Unit =
alterBrokerConfigs(brokerId, newValue, OpType.SET)
+ private def deleteBrokerConfigs(brokerId: String): Unit =
alterBrokerConfigs(brokerId, 0, OpType.DELETE)
+ private def alterBrokerConfigs(brokerId: String, newValue: Long, op:
OpType): Unit = {
+ if (isKRaftTest()) {
+ val admin = createAdminClient()
+ try {
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, brokerId)
+ val configOp = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString), op)
+ val configOp2 = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString), op)
+ val configOp3 = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG,
newValue.toString), op)
+ val configOps = List(configOp, configOp2, configOp3).asJavaCollection
+ admin.incrementalAlterConfigs(Map(
+ resource -> configOps,
+ ).asJava).all.get
+ } finally {
+ admin.close()
+ }
+ } else {
+ val newProps = new Properties()
+ if (op == OpType.SET) {
+ newProps.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString)
+ newProps.put(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString)
+
newProps.put(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG,
newValue.toString)
+ }
+ val brokerIdOption = if (brokerId != "") Option(brokerId.toInt) else None
+ adminZkClient.changeBrokerConfig(brokerIdOption, newProps)
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testBrokerIdConfigChange(quorum: String): Unit = {
+ val newValue: Long = 100000L
+ val brokerId: String = this.brokers.head.config.brokerId.toString
+ setBrokerConfigs(brokerId, newValue)
+ for (b <- this.brokers) {
+ val value = if (b.config.brokerId.toString == brokerId) newValue else
QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT
+ TestUtils.retry(10000) {
+ assertEquals(value, b.quotaManagers.leader.upperBound)
+ assertEquals(value, b.quotaManagers.follower.upperBound)
+ assertEquals(value, b.quotaManagers.alterLogDirs.upperBound)
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testBrokerIdConfigChangeAndDelete(quorum: String): Unit = {
+ val newValue: Long = 100000L
+ val brokerId: String = this.brokers.head.config.brokerId.toString
+ setBrokerConfigs(brokerId, newValue)
+ for (b <- this.brokers) {
+ val value = if (b.config.brokerId.toString == brokerId) newValue else
QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT
+ TestUtils.retry(10000) {
+ assertEquals(value, b.quotaManagers.leader.upperBound)
+ assertEquals(value, b.quotaManagers.follower.upperBound)
+ assertEquals(value, b.quotaManagers.alterLogDirs.upperBound)
+ }
+ }
+ deleteBrokerConfigs(brokerId)
+ for (b <- this.brokers) {
+ TestUtils.retry(10000) {
+ assertEquals(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT,
b.quotaManagers.leader.upperBound)
+ assertEquals(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT,
b.quotaManagers.follower.upperBound)
+ assertEquals(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT,
b.quotaManagers.alterLogDirs.upperBound)
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDefaultBrokerIdConfigChange(quorum: String): Unit = {
Review Comment:
ditto
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -472,6 +472,131 @@ class DynamicConfigChangeTest extends
KafkaServerTestHarness {
}
}
+ private def setBrokerConfigs(brokerId: String, newValue: Long): Unit =
alterBrokerConfigs(brokerId, newValue, OpType.SET)
+ private def deleteBrokerConfigs(brokerId: String): Unit =
alterBrokerConfigs(brokerId, 0, OpType.DELETE)
+ private def alterBrokerConfigs(brokerId: String, newValue: Long, op:
OpType): Unit = {
+ if (isKRaftTest()) {
+ val admin = createAdminClient()
+ try {
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, brokerId)
+ val configOp = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString), op)
+ val configOp2 = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString), op)
+ val configOp3 = new AlterConfigOp(new
ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG,
newValue.toString), op)
+ val configOps = List(configOp, configOp2, configOp3).asJavaCollection
+ admin.incrementalAlterConfigs(Map(
+ resource -> configOps,
+ ).asJava).all.get
+ } finally {
+ admin.close()
+ }
+ } else {
+ val newProps = new Properties()
+ if (op == OpType.SET) {
+ newProps.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString)
+ newProps.put(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
newValue.toString)
+
newProps.put(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG,
newValue.toString)
+ }
+ val brokerIdOption = if (brokerId != "") Option(brokerId.toInt) else None
+ adminZkClient.changeBrokerConfig(brokerIdOption, newProps)
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testBrokerIdConfigChange(quorum: String): Unit = {
Review Comment:
It seems `testBrokerIdConfigChange` is a part of
`testBrokerIdConfigChangeAndDelete`? if so, maybe we can remove
`testBrokerIdConfigChange` to make CI happy :)
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -236,22 +236,27 @@ class IpConfigHandler(private val connectionQuotas:
ConnectionQuotas) extends Co
*/
class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
private val quotaManagers: QuotaManagers) extends
ConfigHandler with Logging {
-
def processConfigChanges(brokerId: String, properties: Properties): Unit = {
- def getOrDefault(prop: String): Long = {
- if (properties.containsKey(prop))
- properties.getProperty(prop).toLong
- else
- QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT
- }
if (brokerId == ZooKeeperInternals.DEFAULT_STRING)
brokerConfig.dynamicConfig.updateDefaultConfig(properties)
else if (brokerConfig.brokerId == brokerId.trim.toInt) {
brokerConfig.dynamicConfig.updateBrokerConfig(brokerConfig.brokerId,
properties)
-
quotaManagers.leader.updateQuota(upperBound(getOrDefault(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG).toDouble))
-
quotaManagers.follower.updateQuota(upperBound(getOrDefault(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG).toDouble))
-
quotaManagers.alterLogDirs.updateQuota(upperBound(getOrDefault(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG).toDouble))
}
+ val updatedDynamicBrokerConfigs =
brokerConfig.dynamicConfig.currentDynamicBrokerConfigs
+ val updatedDynamicDefaultConfigs =
brokerConfig.dynamicConfig.currentDynamicDefaultConfigs
+
+ def getOrDefault(prop: String): Long = updatedDynamicBrokerConfigs get
prop match {
+ case Some(value) => value.toLong
+ case None => {
Review Comment:
please remove unnecessary `{`
--
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
