You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kamalcph (via GitHub)" <gi...@apache.org> on 2024/06/05 04:06:26 UTC
[PR] KAFKA-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
kamalcph opened a new pull request, #16203:
URL: https://github.com/apache/kafka/pull/16203
This patch is built on top of #14778
### 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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #16203:
URL: https://github.com/apache/kafka/pull/16203#issuecomment-2148816401
This patch also depends on #16199
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #16203:
URL: https://github.com/apache/kafka/pull/16203#issuecomment-2159792782
> @kamalcph Can you raise a PR against 3.8 branch as I would like those changes to be reviewed in PR as it has conflicts with that branch?
Opened #16275 and #16276 to cherry-pick the KAFKA-15776 feature to the 3.8 branch. PTAL.
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #16203:
URL: https://github.com/apache/kafka/pull/16203#issuecomment-2158633243
@kamalcph Can you raise a PR against 3.8 branch as I would like those changes to be reviewed in PR as it has conflicts with that branch?
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16203:
URL: https://github.com/apache/kafka/pull/16203#discussion_r1629151230
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
Review Comment:
the test will fail if the validation fails
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
+ config.dynamicConfig.updateDefaultConfig(newProps)
+ assertEquals(30000, config.remoteFetchMaxWaitMs)
+
+ // update per broker config
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "10000")
+ config.dynamicConfig.validate(newProps, perBrokerConfig = true)
Review Comment:
the test will fail if the validation fails
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #16203:
URL: https://github.com/apache/kafka/pull/16203#discussion_r1630681652
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,40 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+ assertEquals(500, config.remoteFetchMaxWaitMs)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
+ config.dynamicConfig.updateDefaultConfig(newProps)
+ assertEquals(30000, config.remoteFetchMaxWaitMs)
+
+ // update per broker config
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "10000")
+ config.dynamicConfig.validate(newProps, perBrokerConfig = true)
+ config.dynamicConfig.updateBrokerConfig(0, newProps)
+ assertEquals(10000, config.remoteFetchMaxWaitMs)
+
+ // invalid value "-1"
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "-1")
+ assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = true))
+ assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = false))
+ // invalid value "0"
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "0")
+ assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = true))
Review Comment:
Can you also check `perBrokerConfig = false` for value 0 like you did for negative value -1 in the earlier code?
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #16203:
URL: https://github.com/apache/kafka/pull/16203#discussion_r1628868196
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
+ config.dynamicConfig.updateDefaultConfig(newProps)
+ assertEquals(30000, config.remoteFetchMaxWaitMs)
+
+ // update per broker config
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "10000")
+ config.dynamicConfig.validate(newProps, perBrokerConfig = true)
Review Comment:
nit: assertDoesNotThrow
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
Review Comment:
We might need to assert the default value of `REMOTE_FETCH_MAX_WAIT_MS_PROP` before any change, to make sure the dynamically change takes effect.
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
+ config.dynamicConfig.updateDefaultConfig(newProps)
+ assertEquals(30000, config.remoteFetchMaxWaitMs)
+
+ // update per broker config
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "10000")
+ config.dynamicConfig.validate(newProps, perBrokerConfig = true)
+ config.dynamicConfig.updateBrokerConfig(0, newProps)
+ assertEquals(10000, config.remoteFetchMaxWaitMs)
+
+ // invalid value
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "-1")
Review Comment:
Could we also test `0` for invalid case? Thanks.
##########
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala:
##########
@@ -792,6 +792,36 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
Review Comment:
nit: assertDoesNotThrow
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #16203:
URL: https://github.com/apache/kafka/pull/16203#issuecomment-2158623607
A few test failures that are unrelated to this change, will merge it to trunk and 3.8 branches.
--
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-15776: Support added to update remote.fetch.max.wait.ms dynamically [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd merged PR #16203:
URL: https://github.com/apache/kafka/pull/16203
--
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