opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Signed-off-by: Aman Gautam <amangautam2128@gmail.com>
This commit is contained in:
Aman Gautam
2026-01-20 11:32:15 +05:30
committed by GitHub
parent ee104114ac
commit 2eb537cdaa
1 changed files with 20 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
+20
View File
@@ -216,6 +216,13 @@ public class PopBufferMergeService extends ServiceThread {
}
}
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 @@ public class PopBufferMergeService extends ServiceThread {
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()) {