MINOR: Fix GroupConfigManager.validate() to include all group type defaults (apache#21593)

`GroupConfigManager.validate()` only included `consumer` group defaults
in the combined configs, causing `share` and `streams` group fields to
fall back to hardcoded defaults. This leads to false validation failures
when broker-configured values differ from those defaults.

Fix by using `extractGroupConfigMap()` which covers all group types, and
adding the missing `extractStreamsGroupConfigMap()` to it.

Reviewers: Andrew Schofield <aschofield@confluent.io>

Author: majialoong

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

@@ -76,17 +76,18 @@ public List<String> groupIds() {
     /**
      * Validate the given properties.
      *
-     * @param newGroupConfig                 The new group config.
-     * @param groupCoordinatorConfig         The group coordinator config.
-     * @throws InvalidConfigurationException If validation fails
+     * @param newGroupConfig         The new group config.
+     * @param groupCoordinatorConfig The group coordinator config.
+     * @param shareGroupConfig       The share group config.
+     * @throws InvalidConfigurationException If validation fails.
      */
     public static void validate(
         Properties newGroupConfig,
         GroupCoordinatorConfig groupCoordinatorConfig,
         ShareGroupConfig shareGroupConfig
     ) {
         Properties combinedConfigs = new Properties();
-        combinedConfigs.putAll(groupCoordinatorConfig.extractConsumerGroupConfigMap());
+        combinedConfigs.putAll(groupCoordinatorConfig.extractGroupConfigMap(shareGroupConfig));
         combinedConfigs.putAll(newGroupConfig);
         GroupConfig.validate(combinedConfigs, groupCoordinatorConfig, shareGroupConfig);
     }

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

@@ -655,12 +655,13 @@ protected List<ShareGroupPartitionAssignor> shareGroupAssignors(
     }
 
     /**
-     * Copy the subset of properties that are relevant to consumer group and share group.
+     * Copy the subset of properties that are relevant to consumer group, share group and streams group.
      */
     public Map<String, Integer> extractGroupConfigMap(ShareGroupConfig shareGroupConfig) {
         Map<String, Integer> defaultConfigs = new HashMap<>();
         defaultConfigs.putAll(extractConsumerGroupConfigMap());
         defaultConfigs.putAll(shareGroupConfig.extractShareGroupConfigMap(this));
+        defaultConfigs.putAll(extractStreamsGroupConfigMap());
         return Collections.unmodifiableMap(defaultConfigs);
     }
 
@@ -673,6 +674,17 @@ GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, consumerGroupSessionTimeoutMs(),
             GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, consumerGroupHeartbeatIntervalMs());
     }
 
+    /**
+     * Copy the subset of properties that are relevant to streams group.
+     */
+    public Map<String, Integer> extractStreamsGroupConfigMap() {
+        return Map.of(
+            GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, streamsGroupSessionTimeoutMs(),
+            GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, streamsGroupHeartbeatIntervalMs(),
+            GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, streamsGroupNumStandbyReplicas(),
+            GroupConfig.STREAMS_INITIAL_REBALANCE_DELAY_MS_CONFIG, streamsGroupInitialRebalanceDelayMs());
+    }
+
     /**
      * The number of threads or event loops running.
      */

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

@@ -17,13 +17,16 @@
 
 package org.apache.kafka.coordinator.group;
 
+import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.errors.InvalidRequestException;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -32,6 +35,7 @@
 import static org.apache.kafka.coordinator.group.GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG;
 import static org.apache.kafka.coordinator.group.GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG;
 import static org.apache.kafka.coordinator.group.GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -81,11 +85,44 @@ public void testUpdateGroupConfig() {
         assertEquals(6000, config.getInt(CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG));
     }
 
+    @Test
+    public void testValidateUsesAllGroupTypeDefaults() {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, 46000);
+        configs.put(GroupCoordinatorConfig.STREAMS_GROUP_SESSION_TIMEOUT_MS_CONFIG, 46000);
+
+        GroupCoordinatorConfig groupCoordinatorConfig = createGroupCoordinatorConfig(configs);
+        ShareGroupConfig shareGroupConfig = createShareGroupConfig();
+
+        Properties newGroupConfig = new Properties();
+        newGroupConfig.put(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, "2");
+
+        assertDoesNotThrow(() ->
+            GroupConfigManager.validate(newGroupConfig, groupCoordinatorConfig, shareGroupConfig));
+    }
+
     public static GroupConfigManager createConfigManager() {
         Map<String, String> defaultConfig = new HashMap<>();
         defaultConfig.put(CONSUMER_SESSION_TIMEOUT_MS_CONFIG, String.valueOf(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_DEFAULT));
         defaultConfig.put(CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, String.valueOf(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_DEFAULT));
         defaultConfig.put(SHARE_RECORD_LOCK_DURATION_MS_CONFIG, String.valueOf(ShareGroupConfig.SHARE_GROUP_RECORD_LOCK_DURATION_MS_DEFAULT));
         return new GroupConfigManager(defaultConfig);
     }
+
+    private static GroupCoordinatorConfig createGroupCoordinatorConfig(Map<String, Object> overrides) {
+        Map<String, Object> configs = new HashMap<>(overrides);
+        return new GroupCoordinatorConfig(new AbstractConfig(
+            GroupCoordinatorConfig.CONFIG_DEF,
+            configs,
+            false
+        ));
+    }
+
+    private static ShareGroupConfig createShareGroupConfig() {
+        return new ShareGroupConfig(new AbstractConfig(
+            Utils.mergeConfigs(Arrays.asList(ShareGroupConfig.CONFIG_DEF, GroupCoordinatorConfig.CONFIG_DEF)),
+            new HashMap<>(),
+            false
+        ));
+    }
 }