squah-confluent commented on code in PR #22935:
URL: https://github.com/apache/kafka/pull/22935#discussion_r3708450906
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
+ }
+
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldHeartbeatInterval(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid heartbeat interval range is 5000 to 15000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "16000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "50000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
Review Comment:
This test passes regardless of whether the new session timeout is checked
against the unevaluated or evaluated heartbeat interval.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
Review Comment:
Could you insert a comment explaining that the heartbeat must be less than
the session timeout?
```
// The heartbeat interval must be less than the session timeout to be valid.
```
or similar.
Applies to `testGroupConfigChangeUsesEvaluatedOldHeartbeatInterval` too.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
Review Comment:
Can we rename this to `testInvalidGroupConfigName` to distinguish it from
the new `testInvalid(Un)changedGroupConfig` tests?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java:
##########
@@ -628,6 +628,32 @@ public void testNotValidatedWhenNotConfigured() {
assertDoesNotThrow(() -> GroupConfig.validate(Map.of(),
groupCoordinatorConfig, createShareGroupConfig()));
}
+ @Test
+ public void testAlterValidationIgnoresUnchangedOutOfRangeValue() {
Review Comment:
The `Alter` can be removed from the name once the other `validate` overload
is removed.
Applies to `testAlterValidationRejectsChangedOutOfRangeValue` too.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java:
##########
@@ -628,6 +628,32 @@ public void testNotValidatedWhenNotConfigured() {
assertDoesNotThrow(() -> GroupConfig.validate(Map.of(),
groupCoordinatorConfig, createShareGroupConfig()));
}
+ @Test
+ public void testAlterValidationIgnoresUnchangedOutOfRangeValue() {
+ // The current valid session timeout range is 45000 to 60000 ms.
+ Map<String, String> oldGroupConfig = Map.of(
+ GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "70000"
+ );
+ Map<String, String> newConfig = new HashMap<>(oldGroupConfig);
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG,
"6000");
+
+ assertDoesNotThrow(() -> GroupConfig.validate(
+ newConfig, oldGroupConfig, "group",
createGroupCoordinatorConfig(), createShareGroupConfig()));
+ }
+
+ @Test
+ public void testAlterValidationRejectsChangedOutOfRangeValue() {
+ // The current valid session timeout range is 45000 to 60000 ms.
+ Map<String, String> oldGroupConfig = Map.of(
+ GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "50000"
+ );
+ Map<String, String> newConfig = new HashMap<>(oldGroupConfig);
+ newConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "70000");
+
+ assertThrows(InvalidConfigurationException.class, () ->
GroupConfig.validate(
+ newConfig, oldGroupConfig, "group",
createGroupCoordinatorConfig(), createShareGroupConfig()));
+ }
+
Review Comment:
`GroupConfigTest` is missing the equivalents for
`testInvalidChangedGroupConfigIsRejected` and
`testInvalidUnchangedGroupConfigIsAccepted`?
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
+ }
+
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldHeartbeatInterval(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid heartbeat interval range is 5000 to 15000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "16000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "50000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
+ }
+
+ @Test
+ def testInvalidChangedGroupConfigIsStillRejected(): Unit = {
Review Comment:
The "Still" in the name is superfluous.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
+ }
+
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldHeartbeatInterval(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid heartbeat interval range is 5000 to 15000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "16000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "50000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
+ }
+
+ @Test
+ def testInvalidChangedGroupConfigIsStillRejected(): Unit = {
Review Comment:
`testInvalidUnchangedGroupConfigIsAccepted` is missing?
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
Review Comment:
Could you insert comments for the valid heartbeat interval too?
```
// The current valid heartbeat interval range is 5000 to 15000 ms.
```
Applies to `newConfig` values in all tests, both in this file and in
`GroupConfigTest`.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
Review Comment:
`GroupConfigTest` should also have a copy of these
uses-valuated-old-<session timeout/heartbeat interval> tests.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -184,6 +184,44 @@ class ControllerConfigurationValidatorTest {
validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap())
}
+ @Test
+ def testGroupConfigChangeUsesEvaluatedOldSessionTimeout(): Unit = {
+ val oldConfig = new util.TreeMap[String, String]()
+ // The current valid session timeout range is 45000 to 60000 ms.
+ oldConfig.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "90000")
+
+ val newConfig = new util.TreeMap[String, String](oldConfig)
+ newConfig.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000")
+
+ validator.validate(new ConfigResource(GROUP, "group"), newConfig,
oldConfig)
Review Comment:
This test passes regardless of whether the new heartbeat interval is checked
against the unevaluated or evaluated session timeout.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
Review Comment:
All the usages in `GroupConfigTest` are in methods intended to test
`GroupConfig.validate`. They exist _because_ `GroupConfig.validate` exists and
cannot be used to justify keeping this overload.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
