You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "harryfallows (via GitHub)" <gi...@apache.org> on 2024/06/11 11:33:09 UTC
[PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
harryfallows opened a new pull request, #16280:
URL: https://github.com/apache/kafka/pull/16280
Fixes issue with dynamic broker configurations where entity default value is never applied. This is a copy of https://github.com/apache/kafka/pull/8906 which was abandoned.
A new unit test has been added to test the BrokerConfigHandler.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639672823
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Do you mean do something similar to the `TopicConfigHandler` `processConfigChanges` method where the update is done in the helper function? In this case we could call `updateQuota` only if the value or defaultValue has changed.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639672823
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Do you mean do something similar to the `TopicConfigHandler` `processConfigChanges` method where the update is done in the helper function? In this case we could call `updateQuota` only if the value or defaultValue has changed, which could be nice. The alternative would be doing the same logic but keeping the `getOrDefault` method and adding a few if statements at the end, which won't be the nicest.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16280:
URL: https://github.com/apache/kafka/pull/16280
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2172345147
@harryfallows thanks for your contribution!
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639845590
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
I've updated it, WDYT? Sorry if my code is dodgy, this is my first time writing Scala.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2165300162
Hey @mimaison, @chia7712, @satishd would it be possible to get a test rerun and a review of this please? Thank you
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2171472423
thanks for the review, I have addressed the comments!
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1640131270
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Sorry that we need `QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT`, since users may remove both "dynamic" value and "default" value. In that scenario, we need to set `QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT` to quota. Please ignore my previous comment.
BTW, both `currentDynamicDefaultConfigs` and `currentDynamicBrokerConfigs` do the deep copy. Could we reuse them?
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2172694775
> @harryfallows thanks for your contribution!
thanks for the reviews!
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1640385788
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Ok, I have reverted my previous change and removed the duplicated deep copies.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639672823
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Do you mean do something similar to the `TopicConfigHandler` `processConfigChanges` method where the update is done in the helper function? In this case we could call `updateQuota` only if the value or defaultValue has changed. Doing the same logic but keeping the `getOrDefault` method will just be a few if statements at the end, which won't be the nicest.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639460364
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Maybe we don't need to update quota if `properties` has no related configs. The default value `QUOTA_BYTES_PER_SECOND_DEFAULT` is equal to "no quota limit". WDYT?
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639672823
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
Do you mean do something similar to the `TopicConfigHandler` `processConfigChanges` method where the update is done in the helper function? In this case we could call `updateQuota` if the value or defaultValue has changed.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1639750736
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -239,19 +239,24 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
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
+ brokerConfig.dynamicConfig.currentDynamicBrokerConfigs get prop match {
Review Comment:
> In this case we could call updateQuota only if the value or defaultValue has changed, which could be nice.
I prefer this approach.
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2162378277
The failing test is a streams one, and it passes locally so I think it's a flake.
```
hfallows@hfallows:~/kafka$ ./gradlew -PscalaVersion=2.12 :streams:streams-scala:test --profile --continue -PkeepAliveMode=session -PtestLoggingEvents=started,passed,skipped,failed -PignoreFailures=true -PmaxParallelForks=2 -PmaxTestRetries=1 -PmaxTestRetryFailures=10
Starting a Gradle Daemon (subsequent builds will be faster)
...
BUILD SUCCESSFUL in 4m 53s
83 actionable tasks: 17 executed, 66 up-to-date
See the profiling report at: file:///home/hfallows/kafka/build/reports/profile/profile-2024-06-12-09-00-08.html
A fine-grained performance profile is available: use the --scan option.
```
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16280:
URL: https://github.com/apache/kafka/pull/16280#discussion_r1640574384
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -472,6 +472,80 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
}
}
+ private def alterBrokerConfigs(brokerId: String, newValue: java.lang.Long): Unit = {
+ if (isKRaftTest()) {
+ val admin = createAdminClient()
+ try {
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, brokerId)
+ val configEntry = new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, newValue.toString)
+ val configEntry2 = new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, newValue.toString)
+ val configEntry3 = new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, newValue.toString)
+ val config = new Config(List(configEntry, configEntry2, configEntry3).asJavaCollection)
+ admin.alterConfigs(Map(
Review Comment:
Could you please use `incrementalAlterConfigs` instead of deprecated `alterConfigs`?
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -472,6 +472,80 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
}
}
+ private def alterBrokerConfigs(brokerId: String, newValue: java.lang.Long): Unit = {
+ if (isKRaftTest()) {
+ val admin = createAdminClient()
+ try {
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, brokerId)
+ val configEntry = new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, newValue.toString)
+ val configEntry2 = new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, newValue.toString)
+ val configEntry3 = new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, newValue.toString)
+ val config = new Config(List(configEntry, configEntry2, configEntry3).asJavaCollection)
+ admin.alterConfigs(Map(
+ resource -> config,
+ ).asJava).all.get
+ } finally {
+ admin.close()
+ }
+ } else {
+ val newProps = new Properties()
+ 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: java.lang.Long = 100000L
+ val brokerId: String = this.brokers.head.config.brokerId.toString
+ alterBrokerConfigs(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 testDefaultBrokerIdConfigChange(quorum: String): Unit = {
+ val newValue: java.lang.Long = 100000L
+ val brokerId: String = ""
+ alterBrokerConfigs(brokerId, newValue)
+ for (b <- this.brokers) {
+ TestUtils.retry(10000) {
+ assertEquals(newValue, b.quotaManagers.leader.upperBound)
+ assertEquals(newValue, b.quotaManagers.follower.upperBound)
+ assertEquals(newValue, b.quotaManagers.alterLogDirs.upperBound)
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDefaultAndBrokerIdConfigChange(quorum: String): Unit = {
+ val newValue: java.lang.Long = 100000L
+ val brokerId: String = this.brokers.head.config.brokerId.toString
+ alterBrokerConfigs(brokerId, newValue)
+ val newDefaultValue: java.lang.Long = 200000L
+ alterBrokerConfigs("", newDefaultValue)
+ for (b <- this.brokers) {
+ val value = if (b.config.brokerId.toString == brokerId) newValue else newDefaultValue
+ TestUtils.retry(10000) {
+ assertEquals(value, b.quotaManagers.leader.upperBound)
+ assertEquals(value, b.quotaManagers.follower.upperBound)
+ assertEquals(value, b.quotaManagers.alterLogDirs.upperBound)
+ }
+ }
+ }
+
Review Comment:
Could you please add a test that unset both dynamic and default value?
##########
core/src/main/scala/kafka/server/ConfigHandler.scala:
##########
@@ -236,22 +236,29 @@ 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 redundant `{}`
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -472,6 +472,80 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
}
}
+ private def alterBrokerConfigs(brokerId: String, newValue: java.lang.Long): Unit = {
Review Comment:
It seems `java.lang.Long` can be superseded by `Long`
--
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
Re: [PR] KAFKA-10190: Set dynamic broker configs for entity default [kafka]
Posted by "harryfallows (via GitHub)" <gi...@apache.org>.
harryfallows commented on PR #16280:
URL: https://github.com/apache/kafka/pull/16280#issuecomment-2169405561
thanks @chia7712, I've addressed your comments
--
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