MINOR: Include numeric bound in group config validation errors (#22185)

When a dynamic group config value is out of range, the error message
referenced the broker-level config name that defines the bound (e.g.
`group.consumer.min.heartbeat.interval.ms`). Operators had to look up
the broker config to know which value would be accepted.

This patch updates the `validateIntRange`, `validateIntMin`, and
`validateIntMax` helpers in `GroupConfig` to include the numeric bound
directly. For example, setting `consumer.heartbeat.interval.ms` below
the minimum now reports `consumer.heartbeat.interval.ms must be greater
than or equal to 5`.

A new parameterized test `testValidationErrorMessageIncludesBound`
covers both directions for each range-bounded config and the max-only
and min-only checks.

Reviewers: Andrew Schofield <aschofield@confluent.io>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: dajac

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -1154,7 +1154,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
 
     assertFutureThrows(classOf[InvalidConfigurationException],
       alterResult.values.get(groupResource),
-      "consumer.session.timeout.ms must be greater than or equal to group.consumer.min.session.timeout.ms")
+      "consumer.session.timeout.ms must be in the range 45000 to 60000 inclusive.")
   }
 
   @Test

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java

@@ -446,119 +446,92 @@ private static void validateValues(
             parsed,
             CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.consumerGroupMinHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.consumerGroupMaxHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.consumerGroupMaxHeartbeatIntervalMs()
         );
         validateIntRange(
             parsed,
             CONSUMER_SESSION_TIMEOUT_MS_CONFIG,
             groupCoordinatorConfig.consumerGroupMinSessionTimeoutMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG,
-            groupCoordinatorConfig.consumerGroupMaxSessionTimeoutMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG
+            groupCoordinatorConfig.consumerGroupMaxSessionTimeoutMs()
         );
         validateIntRange(
             parsed,
             CONSUMER_ASSIGNMENT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.consumerGroupMinAssignmentIntervalMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.consumerGroupMaxAssignmentIntervalMs(),
-            GroupCoordinatorConfig.CONSUMER_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.consumerGroupMaxAssignmentIntervalMs()
         );
 
         // Share group configs.
         validateIntRange(
             parsed,
             SHARE_HEARTBEAT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.shareGroupMinHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.shareGroupMaxHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.shareGroupMaxHeartbeatIntervalMs()
         );
         validateIntRange(
             parsed,
             SHARE_SESSION_TIMEOUT_MS_CONFIG,
             groupCoordinatorConfig.shareGroupMinSessionTimeoutMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG,
-            groupCoordinatorConfig.shareGroupMaxSessionTimeoutMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG
+            groupCoordinatorConfig.shareGroupMaxSessionTimeoutMs()
         );
         validateIntRange(
             parsed,
             SHARE_RECORD_LOCK_DURATION_MS_CONFIG,
             shareGroupConfig.shareGroupMinRecordLockDurationMs(),
-            ShareGroupConfig.SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_CONFIG,
-            shareGroupConfig.shareGroupMaxRecordLockDurationMs(),
-            ShareGroupConfig.SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_CONFIG
+            shareGroupConfig.shareGroupMaxRecordLockDurationMs()
         );
         validateIntRange(
             parsed,
             SHARE_DELIVERY_COUNT_LIMIT_CONFIG,
             shareGroupConfig.shareGroupMinDeliveryCountLimit(),
-            ShareGroupConfig.SHARE_GROUP_MIN_DELIVERY_COUNT_LIMIT_CONFIG,
-            shareGroupConfig.shareGroupMaxDeliveryCountLimit(),
-            ShareGroupConfig.SHARE_GROUP_MAX_DELIVERY_COUNT_LIMIT_CONFIG
+            shareGroupConfig.shareGroupMaxDeliveryCountLimit()
         );
         validateIntRange(
             parsed,
             SHARE_PARTITION_MAX_RECORD_LOCKS_CONFIG,
             shareGroupConfig.shareGroupMinPartitionMaxRecordLocks(),
-            ShareGroupConfig.SHARE_GROUP_MIN_PARTITION_MAX_RECORD_LOCKS_CONFIG,
-            shareGroupConfig.shareGroupMaxPartitionMaxRecordLocks(),
-            ShareGroupConfig.SHARE_GROUP_MAX_PARTITION_MAX_RECORD_LOCKS_CONFIG
+            shareGroupConfig.shareGroupMaxPartitionMaxRecordLocks()
         );
         validateIntRange(
             parsed,
             SHARE_ASSIGNMENT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.shareGroupMinAssignmentIntervalMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.shareGroupMaxAssignmentIntervalMs(),
-            GroupCoordinatorConfig.SHARE_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.shareGroupMaxAssignmentIntervalMs()
         );
 
         // Streams group configs.
         validateIntRange(
             parsed,
             STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.streamsGroupMinHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMaxHeartbeatIntervalMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.streamsGroupMaxHeartbeatIntervalMs()
         );
         validateIntRange(
             parsed,
             STREAMS_SESSION_TIMEOUT_MS_CONFIG,
             groupCoordinatorConfig.streamsGroupMinSessionTimeoutMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMaxSessionTimeoutMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG
+            groupCoordinatorConfig.streamsGroupMaxSessionTimeoutMs()
         );
         validateIntMax(
             parsed,
             STREAMS_NUM_STANDBY_REPLICAS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMaxNumStandbyReplicas(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MAX_STANDBY_REPLICAS_CONFIG
+            groupCoordinatorConfig.streamsGroupMaxNumStandbyReplicas()
         );
         validateIntRange(
             parsed,
             STREAMS_ASSIGNMENT_INTERVAL_MS_CONFIG,
             groupCoordinatorConfig.streamsGroupMinAssignmentIntervalMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMaxAssignmentIntervalMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.streamsGroupMaxAssignmentIntervalMs()
         );
         validateIntMin(
             parsed,
             STREAMS_TASK_OFFSET_INTERVAL_MS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMinTaskOffsetIntervalMs(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_CONFIG
+            groupCoordinatorConfig.streamsGroupMinTaskOffsetIntervalMs()
         );
         validateIntMax(
             parsed,
             STREAMS_NUM_WARMUP_REPLICAS_CONFIG,
-            groupCoordinatorConfig.streamsGroupMaxWarmupReplicas(),
-            GroupCoordinatorConfig.STREAMS_GROUP_MAX_WARMUP_REPLICAS_CONFIG
+            groupCoordinatorConfig.streamsGroupMaxWarmupReplicas()
         );
 
         // Cross-field validations: session timeout must be greater than heartbeat interval.
@@ -601,16 +574,12 @@ private static void validateIntRange(
         Map<String, Object> parsed,
         String key,
         int min,
-        String minConfigName,
-        int max,
-        String maxConfigName
+        int max
     ) {
         if (!parsed.containsKey(key)) return;
         int value = (Integer) parsed.get(key);
-        if (value < min)
-            throw new InvalidConfigurationException(key + " must be greater than or equal to " + minConfigName);
-        if (value > max)
-            throw new InvalidConfigurationException(key + " must be less than or equal to " + maxConfigName);
+        if (value < min || value > max)
+            throw new InvalidConfigurationException(key + " must be in the range " + min + " to " + max + " inclusive.");
     }
 
     /**
@@ -620,13 +589,12 @@ private static void validateIntRange(
     private static void validateIntMax(
         Map<String, Object> parsed,
         String key,
-        int max,
-        String maxConfigName
+        int max
     ) {
         if (!parsed.containsKey(key)) return;
         int value = (Integer) parsed.get(key);
         if (value > max)
-            throw new InvalidConfigurationException(key + " must be less than or equal to " + maxConfigName);
+            throw new InvalidConfigurationException(key + " must be less than or equal to " + max);
     }
 
     /**
@@ -636,13 +604,12 @@ private static void validateIntMax(
     private static void validateIntMin(
         Map<String, Object> parsed,
         String key,
-        int min,
-        String minConfigName
+        int min
     ) {
         if (!parsed.containsKey(key)) return;
         int value = (Integer) parsed.get(key);
         if (value < min)
-            throw new InvalidConfigurationException(key + " must be greater than or equal to " + minConfigName);
+            throw new InvalidConfigurationException(key + " must be greater than or equal to " + min);
     }
 
     /**

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java

@@ -353,6 +353,77 @@ private void doTestValidProps(Map<String, String> props) {
         assertDoesNotThrow(() -> GroupConfig.validate(props, createGroupCoordinatorConfig(), createShareGroupConfig()));
     }
 
+    private static Stream<Arguments> outOfRangeValuesAndExpectedMessages() {
+        return Stream.of(
+            // Consumer group configs.
+            Arguments.of(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "1",
+                "consumer.heartbeat.interval.ms must be in the range 5 to 15000 inclusive."),
+            Arguments.of(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "20000",
+                "consumer.heartbeat.interval.ms must be in the range 5 to 15000 inclusive."),
+            Arguments.of(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "1",
+                "consumer.session.timeout.ms must be in the range 45 to 60000 inclusive."),
+            Arguments.of(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "70000",
+                "consumer.session.timeout.ms must be in the range 45 to 60000 inclusive."),
+            Arguments.of(GroupConfig.CONSUMER_ASSIGNMENT_INTERVAL_MS_CONFIG, "500",
+                "consumer.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.CONSUMER_ASSIGNMENT_INTERVAL_MS_CONFIG, "20000",
+                "consumer.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+
+            // Share group configs.
+            Arguments.of(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "1",
+                "share.heartbeat.interval.ms must be in the range 5 to 15000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "20000",
+                "share.heartbeat.interval.ms must be in the range 5 to 15000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "1",
+                "share.session.timeout.ms must be in the range 45 to 60000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "70000",
+                "share.session.timeout.ms must be in the range 45 to 60000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "10000",
+                "share.record.lock.duration.ms must be in the range 15000 to 60000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "70000",
+                "share.record.lock.duration.ms must be in the range 15000 to 60000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_DELIVERY_COUNT_LIMIT_CONFIG, "11",
+                "share.delivery.count.limit must be in the range 2 to 10 inclusive."),
+            Arguments.of(GroupConfig.SHARE_PARTITION_MAX_RECORD_LOCKS_CONFIG, "11000",
+                "share.partition.max.record.locks must be in the range 100 to 10000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_ASSIGNMENT_INTERVAL_MS_CONFIG, "500",
+                "share.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.SHARE_ASSIGNMENT_INTERVAL_MS_CONFIG, "20000",
+                "share.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+
+            // Streams group configs.
+            Arguments.of(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "1000",
+                "streams.heartbeat.interval.ms must be in the range 5000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "20000",
+                "streams.heartbeat.interval.ms must be in the range 5000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "1",
+                "streams.session.timeout.ms must be in the range 45000 to 60000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "70000",
+                "streams.session.timeout.ms must be in the range 45000 to 60000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, "5",
+                "streams.num.standby.replicas must be less than or equal to 2"),
+            Arguments.of(GroupConfig.STREAMS_ASSIGNMENT_INTERVAL_MS_CONFIG, "500",
+                "streams.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_ASSIGNMENT_INTERVAL_MS_CONFIG, "20000",
+                "streams.assignment.interval.ms must be in the range 1000 to 15000 inclusive."),
+            Arguments.of(GroupConfig.STREAMS_TASK_OFFSET_INTERVAL_MS_CONFIG, "1000",
+                "streams.task.offset.interval.ms must be greater than or equal to 15000"),
+            Arguments.of(GroupConfig.STREAMS_NUM_WARMUP_REPLICAS_CONFIG, "50",
+                "streams.num.warmup.replicas must be less than or equal to 20")
+        );
+    }
+
+    @ParameterizedTest(name = "testValidationErrorMessageIncludesBound[{0}={1}]")
+    @MethodSource("outOfRangeValuesAndExpectedMessages")
+    public void testValidationErrorMessageIncludesBound(String key, String value, String expectedMessage) {
+        var props = Map.of(key, value);
+        var exception = assertThrows(
+            InvalidConfigurationException.class,
+            () -> GroupConfig.validate(props, createGroupCoordinatorConfig(), createShareGroupConfig())
+        );
+        assertEquals(expectedMessage, exception.getMessage());
+    }
+
     @Test
     public void testFromPropsWithDefaultValue() {
         Map<String, String> defaultValue = new HashMap<>();

tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java

@@ -316,15 +316,15 @@ public void testAlterStreamsGroupSessionTimeout() {
             "--entity-name", "group",
             "--alter", "--add-config", "streams.session.timeout.ms=1"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.session.timeout.ms must be greater than or equal to group.streams.min.session.timeout.ms"));
+        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.session.timeout.ms must be in the range 45000 to 60000 inclusive."));
 
         // Should fail to set above max
         command = Stream.concat(quorumArgs(), Stream.of(
             "--entity-type", "groups",
             "--entity-name", "group",
             "--alter", "--add-config", "streams.session.timeout.ms=100000"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.session.timeout.ms must be less than or equal to group.streams.max.session.timeout.ms"));
+        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.session.timeout.ms must be in the range 45000 to 60000 inclusive."));
     }
 
     @ClusterTest(serverProperties = {
@@ -347,15 +347,15 @@ public void testAlterStreamsGroupHeartbeatInterval() {
             "--entity-name", "group",
             "--alter", "--add-config", "streams.heartbeat.interval.ms=1"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.heartbeat.interval.ms must be greater than or equal to group.streams.min.heartbeat.interval.ms"));
+        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.heartbeat.interval.ms must be in the range 5000 to 55000 inclusive."));
 
         // Should fail to set above max
         command = Stream.concat(quorumArgs(), Stream.of(
             "--entity-type", "groups",
             "--entity-name", "group",
             "--alter", "--add-config", "streams.heartbeat.interval.ms=100000"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.heartbeat.interval.ms must be less than or equal to group.streams.max.heartbeat.interval.ms"));
+        assertTrue(message.contains("org.apache.kafka.common.errors.InvalidConfigurationException: streams.heartbeat.interval.ms must be in the range 5000 to 55000 inclusive."));
 
         // Should fail to set above session timeout
         command = Stream.concat(quorumArgs(), Stream.of(
@@ -398,7 +398,7 @@ public void testAlterStreamsGroupNumOfStandbyReplicas() {
             "--entity-name", "group",
             "--alter", "--add-config", "streams.num.standby.replicas=3"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("streams.num.standby.replicas must be less than or equal to group.streams.max.standby.replicas"));
+        assertTrue(message.contains("streams.num.standby.replicas must be less than or equal to 2"));
     }
 
     @ClusterTest
@@ -432,7 +432,7 @@ public void testAlterStreamsGroupTaskOffsetInterval() {
             "--entity-name", "group",
             "--alter", "--add-config", "streams.task.offset.interval.ms=1"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("streams.task.offset.interval.ms must be greater than or equal to group.streams.min.task.offset.interval.ms"));
+        assertTrue(message.contains("streams.task.offset.interval.ms must be greater than or equal to 15000"));
 
     }
 
@@ -467,7 +467,7 @@ public void testAlterStreamsGroupNumWarmupReplicas() {
             "--entity-name", "group",
             "--alter", "--add-config", "streams.num.warmup.replicas=25"));
         message = captureStandardErr(run(command));
-        assertTrue(message.contains("streams.num.warmup.replicas must be less than or equal to group.streams.max.warmup.replicas"));
+        assertTrue(message.contains("streams.num.warmup.replicas must be less than or equal to 20"));
     }
 
     private void verifyGroupConfigUpdate(List<String> alterOpts) throws Exception {