KAFKA-20288 Remove orphaned groups from GroupConfigManager (#21758)

# Description
`GroupConfigManager` stores the dynamic configs of groups. However, when
all dynamic configs are removed the group id is not removed from the
manager. This patch removes the orphan when the update group config is
empty.

# Test
`testGroupIsRemovedWhenDynamicConfigsAreRemoved` checks whether the
orphan group exists after updating using an empty config.

Reviewers: Sean Quah <squah@confluent.io>, David Jacot
<djacot@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

Author: jack2012aa

core/src/main/java/kafka/server/share/SharePartition.java

@@ -3040,12 +3040,9 @@ private long startOffsetDuringInitialization(long partitionDataStartOffset) {
         if (partitionDataStartOffset != PartitionFactory.UNINITIALIZED_START_OFFSET) {
             return partitionDataStartOffset;
         }
-        ShareGroupAutoOffsetResetStrategy offsetResetStrategy;
-        if (groupConfigManager.groupConfig(groupId).isPresent()) {
-            offsetResetStrategy = groupConfigManager.groupConfig(groupId).get().shareAutoOffsetReset();
-        } else {
-            offsetResetStrategy = GroupConfig.defaultShareAutoOffsetReset();
-        }
+        ShareGroupAutoOffsetResetStrategy offsetResetStrategy = groupConfigManager.groupConfig(groupId)
+            .map(GroupConfig::shareAutoOffsetReset)
+            .orElseGet(GroupConfig::defaultShareAutoOffsetReset);
 
         if (offsetResetStrategy.type() == ShareGroupAutoOffsetResetStrategy.StrategyType.LATEST) {
             return offsetForLatestTimestamp(topicIdPartition, replicaManager, leaderEpoch);

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

@@ -64,6 +64,11 @@ public void updateGroupConfig(String groupId, Properties newGroupConfig) {
             throw new InvalidRequestException("Group name can't be empty.");
         }
 
+        if (newGroupConfig.isEmpty()) {
+            configMap.remove(groupId);
+            return;
+        }
+
         // Evaluate ensures configs respect broker-level bounds. For the Admin API path,
         // values are pre-validated so this is effectively a no-op. For the broker startup
         // path, configs from metadata may need evaluation if bounds have changed.

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

@@ -115,6 +115,21 @@ public void testClampWithCustomBrokerBounds() {
         assertEquals(49000, configManager.groupConfig(groupId).get().getInt(CONSUMER_SESSION_TIMEOUT_MS_CONFIG));
     }
 
+    @Test
+    public void testGroupIsRemovedWhenDynamicConfigsAreRemoved() {
+        String groupId1 = "foo";
+        String groupId2 = "bar";
+        Properties props = new Properties();
+        props.put(CONSUMER_SESSION_TIMEOUT_MS_CONFIG, 50000);
+        configManager.updateGroupConfig(groupId1, props);
+        configManager.updateGroupConfig(groupId2, props);
+        assertTrue(configManager.groupIds().contains(groupId1));
+
+        configManager.updateGroupConfig(groupId1, new Properties());
+        assertFalse(configManager.groupIds().contains(groupId1));
+        assertTrue(configManager.groupIds().contains(groupId2));
+    }
+
     public static GroupConfigManager createConfigManager() {
         return createConfigManager(new HashMap<>());
     }

server/src/test/java/org/apache/kafka/server/config/DynamicConfigTest.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.config;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.admin.ListConfigResourcesOptions;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.test.TestUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class DynamicConfigTest {
+    @ClusterTest
+    public void testGroupIsRemovedWhenDynamicConfigsAreRemoved(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
+        try (Admin admin = clusterInstance.admin()) {
+            var cr = new ConfigResource(ConfigResource.Type.GROUP, "gp");
+            assertEquals(List.of(), admin.listConfigResources(Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions()).all().get());
+
+            // add dynamic config
+            admin.incrementalAlterConfigs(Map.of(cr, List.of(new AlterConfigOp(
+                new ConfigEntry("consumer.session.timeout.ms", "45001"), AlterConfigOp.OpType.SET))))
+                .all()
+                .get();
+            TestUtils.waitForCondition(() -> !admin.listConfigResources(Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions()).all().get().isEmpty(),
+                "Should include a group with dynamic config");
+
+            // remove dynamic config
+            admin.incrementalAlterConfigs(Map.of(cr, List.of(new AlterConfigOp(
+                new ConfigEntry("consumer.session.timeout.ms", null), AlterConfigOp.OpType.DELETE))))
+                .all()
+                .get();
+            TestUtils.waitForCondition(() -> admin.listConfigResources(Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions()).all().get().isEmpty(),
+                "Should not include any group");
+        }
+    }
+}