diff --git a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
index 85bc8e3789..eafb47a89d 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicLong;
 
 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 class ConsumerOffsetManager extends ConfigManager {
     }
 
     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) {
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
index 7bd289a6f1..9fc553409d 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
+++ b/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.ConcurrentHashMap;
 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 class ConsumerOffsetManagerTest {
         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();