diff --git a/broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java b/broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java index ce70b1a820..53e9e2e063 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java @@ -17,6 +17,7 @@ package org.apache.rocketmq.broker.offset; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -110,4 +111,25 @@ public class LmqConsumerOffsetManager extends ConsumerOffsetManager { public void setLmqOffsetTable(ConcurrentHashMap lmqOffsetTable) { this.lmqOffsetTable = lmqOffsetTable; } + + @Override + public void removeOffset(String group) { + if (!MixAll.isLmq(group)) { + super.removeOffset(group); + return; + } + Iterator> it = this.lmqOffsetTable.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry 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 lmq group offset {}", topicAtGroup); + } + } + } + } }