[ISSUE #10181] Remove lock when resetting offset in PopConsumerService (#10250)

Author: lizhimins

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java

@@ -555,16 +555,9 @@ public CompletableFuture<Boolean> revive(PopConsumerRecord record) {
             });
     }
 
-    @SuppressWarnings("StatementWithEmptyBody")
     public void clearCache(String groupId, String topicId, int queueId) {
-        while (!consumerLockService.tryLock(groupId, topicId)) {
-        }
-        try {
-            if (popConsumerCache != null) {
-                popConsumerCache.removeRecords(groupId, topicId, queueId);
-            }
-        } finally {
-            consumerLockService.unlock(groupId, topicId);
+        if (popConsumerCache != null) {
+            popConsumerCache.removeRecords(groupId, topicId, queueId);
         }
     }
 

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -2183,7 +2183,6 @@ private RemotingCommand resetOffsetInner(String topic, String group, int queueId
                 brokerController.getPopInflightMessageCounter().clearInFlightMessageNum(topic, group, entry.getKey());
             }
             if (brokerController.getBrokerConfig().isPopConsumerKVServiceEnable()) {
-                brokerController.getPopConsumerService().clearCache(group, topic, entry.getKey());
                 brokerController.getConsumerOffsetManager().clearPullOffset(group, topic);
             }
             body.getOffsetTable().put(new MessageQueue(topic, brokerName, entry.getKey()), entry.getValue());