[ISSUE #9980] Skip invalid Pop records when consumer group does not exist (#9982)

Signed-off-by: Aman Gautam <[email protected]>

Author: Gautam-aman

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

@@ -216,6 +216,13 @@ private void scanGarbage() {
         }
     }
 
+    private boolean isSubscriptionGroupNotExist(PopCheckPointWrapper pointWrapper) {
+        String group = pointWrapper.getCk().getCId();
+        return brokerController.getSubscriptionGroupManager()
+                .findSubscriptionGroupConfig(group) == null;
+    }
+
+
     private void scan() {
         long startTime = System.currentTimeMillis();
         AtomicInteger count = new AtomicInteger(0);
@@ -225,6 +232,19 @@ private void scan() {
             Map.Entry<String, PopCheckPointWrapper> entry = iterator.next();
             PopCheckPointWrapper pointWrapper = entry.getValue();
 
+            // Skip invalid POP records when consumer group does not exist
+            if (isSubscriptionGroupNotExist(pointWrapper)) {
+                POP_LOGGER.warn(
+                        "[PopBuffer] skip pop record because consumer group not exist, group={}, ck={}",
+                        pointWrapper.getCk().getCId(),
+                        pointWrapper
+                );
+                iterator.remove();
+                counter.decrementAndGet();
+                continue;
+            }
+
+
             // just process offset(already stored at pull thread), or buffer ck(not stored and ack finish)
             if (pointWrapper.isJustOffset() && pointWrapper.isCkStored() || isCkDone(pointWrapper)
                 || isCkDoneForFinish(pointWrapper) && pointWrapper.isCkStored()) {