[ISSUE #9172] Clean pull offset and reset offset when delete subscription group (#9173)

* [ISSUE #9172] Clean pull offset and reset offset when delete subscription group

* [ISSUE #9174] Add a collection of predefined Groups and common checking methods in the MixAll (#9175)

Signed-off-by: ltamber <ltamber12@gmail.com>

* [ISSUE #9177] Fix unstable tests in AdaptiveLockTest.testAdaptiveLock  (#9178)

---------

Signed-off-by: ltamber <ltamber12@gmail.com>
Co-authored-by: ltamber <ltamber12@gmail.com>
Co-authored-by: hqbfz <125714719+3424672656@users.noreply.github.com>

Author: 3 people

broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java

@@ -29,6 +29,7 @@
 
 import com.google.common.base.Strings;
 
+import java.util.function.Function;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerPathConfigHelper;
 import org.apache.rocketmq.common.ConfigManager;
@@ -395,19 +396,29 @@ public boolean loadDataVersion() {
     }
 
     public void removeOffset(final String group) {
-        Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
-        while (it.hasNext()) {
-            Entry<String, ConcurrentMap<Integer, Long>> next = it.next();
-            String topicAtGroup = next.getKey();
-            if (topicAtGroup.contains(group)) {
-                String[] arrays = topicAtGroup.split(TOPIC_GROUP_SEPARATOR);
-                if (arrays.length == 2 && group.equals(arrays[1])) {
-                    it.remove();
-                    removeConsumerOffset(topicAtGroup);
-                    LOG.warn("clean group offset {}", topicAtGroup);
+        Function<Iterator<Entry<String, ConcurrentMap<Integer, Long>>>, Boolean> deleteFunction = it -> {
+            boolean removed = false;
+            while (it.hasNext()) {
+                Entry<String, ConcurrentMap<Integer, Long>> entry = it.next();
+                String topicAtGroup = entry.getKey();
+                if (topicAtGroup.contains(group)) {
+                    String[] arrays = topicAtGroup.split(TOPIC_GROUP_SEPARATOR);
+                    if (arrays.length == 2 && group.equals(arrays[1])) {
+                        it.remove();
+                        removeConsumerOffset(topicAtGroup);
+                        removed = true;
+                    }
                 }
             }
-        }
+            return removed;
+        };
+
+        boolean clearOffset = deleteFunction.apply(this.offsetTable.entrySet().iterator());
+        boolean clearReset = deleteFunction.apply(this.resetOffsetTable.entrySet().iterator());
+        boolean clearPull = deleteFunction.apply(this.pullOffsetTable.entrySet().iterator());
+
+        LOG.info("Consumer offset manager clean group offset, groupName={}, " +
+            "offsetTable={}, resetOffsetTable={}, pullOffsetTable={}", group, clearOffset, clearReset, clearPull);
     }
 
     public void assignResetOffset(String topic, String group, int queueId, long offset) {

broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.broker.offset;
 
 import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.junit.Assert;
 import org.junit.Before;
@@ -27,6 +28,7 @@
 import java.util.concurrent.ConcurrentMap;
 import org.mockito.Mockito;
 
+import static org.apache.rocketmq.broker.offset.ConsumerOffsetManager.TOPIC_GROUP_SEPARATOR;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class ConsumerOffsetManagerTest {
@@ -65,6 +67,18 @@ public void cleanOffsetByTopic_Exist() {
         assertThat(!consumerOffsetManager.getOffsetTable().containsKey(KEY)).isTrue();
     }
 
+    @Test
+    public void removeOffsetByGroupTest() {
+        String topic = "TopicName";
+        String group = "GroupName";
+        Mockito.when(brokerController.getBrokerConfig()).thenReturn(new BrokerConfig());
+        consumerOffsetManager.commitOffset("Commit", group, topic, 0, 100);
+        consumerOffsetManager.assignResetOffset(topic, group, 0, 100);
+        consumerOffsetManager.commitPullOffset("Pull", group, topic, 0, 100);
+        consumerOffsetManager.removeOffset(group);
+        Assert.assertFalse(consumerOffsetManager.getOffsetTable().containsKey(topic + TOPIC_GROUP_SEPARATOR + group));
+    }
+
     @Test
     public void testOffsetPersistInMemory() {
         ConcurrentMap<String, ConcurrentMap<Integer, Long>> offsetTable = consumerOffsetManager.getOffsetTable();