[ISSUE apache#10114] Add removeOffsetByTopicAndGroup method for precise consumer offset cleanup

This commit adds a new method removeOffsetByTopicAndGroup() to ConsumerOffsetManager
that allows users to precisely clean up consumer offset for a specific topic under
a specific consumer group, without removing the entire group or topic.

This addresses the enhancement request in issue apache#10114 where users need the ability
to precisely clean up the consumer offset for a specific topic under a specific
consumer group when a consumer group subscribes to an unintended topic by mistake.

Signed-off-by: Senrian <senrian@github.com>

Author: Senrian

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

@@ -103,6 +103,28 @@ public void cleanOffsetByTopic(String topic) {
         }
     }
 
+    /**
+     * Remove consumer offset for a specific topic under a specific consumer group.
+     * This allows precise cleanup without removing the entire group or topic.
+     *
+     * @param topic the topic to remove offset for
+     * @param group the consumer group to remove offset for
+     */
+    public void removeOffsetByTopicAndGroup(String topic, String group) {
+        if (topic == null || group == null) {
+            LOG.warn("Cannot remove offset: topic or group is null. topic={}, group={}", topic, group);
+            return;
+        }
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        ConcurrentMap<Integer, Long> removed = this.offsetTable.remove(key);
+        if (removed != null) {
+            removeConsumerOffset(key);
+            pullOffsetTable.remove(key);
+            resetOffsetTable.remove(key);
+            LOG.info("Removed consumer offset for topic={}, group={}, offsetCount={}", topic, group, removed.size());
+        }
+    }
+
     public void scanUnsubscribedTopic() {
         Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
         while (it.hasNext()) {