[ISSUE #9980] Skip invalid records when the group is absent in Pop (#9981)

Signed-off-by: terrance.lzm <terrance.lzm@alibaba-inc.com>

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

@@ -501,7 +501,15 @@ public class PopConsumerService extends ServiceThread {
         PopConsumerRecord ackRecord = new PopConsumerRecord(
             popTime, groupId, topicId, queueId, 0, invisibleTime, offset, null);
 
-        this.popConsumerStore.writeRecords(Collections.singletonList(ckRecord));
+        // No need to generate new records when the group does not exist,
+        // because these retry messages will not be consumed by anyone.
+        if (brokerConfig.isPopReviveSkipIfGroupAbsent() &&
+            !brokerController.getSubscriptionGroupManager().containsSubscriptionGroup(groupId)) {
+            log.info("PopConsumerService change invisibility skip, time={}, " +
+                "groupId={}, topicId={}, queueId={}, offset={}", popTime, groupId, topicId, queueId, offset);
+        } else {
+            this.popConsumerStore.writeRecords(Collections.singletonList(ckRecord));
+        }
 
         if (brokerConfig.isEnablePopBufferMerge() && popConsumerCache != null) {
             if (popConsumerCache.deleteRecords(Collections.singletonList(ackRecord)).isEmpty()) {
@@ -519,6 +527,13 @@ public class PopConsumerService extends ServiceThread {
     }
 
     public CompletableFuture<Boolean> revive(PopConsumerRecord record) {
+
+        if (brokerConfig.isPopReviveSkipIfGroupAbsent() &&
+            !brokerController.getSubscriptionGroupManager().containsSubscriptionGroup(record.getGroupId())) {
+            log.info("PopConsumerService skip revive message, record={}", record);
+            return CompletableFuture.completedFuture(true);
+        }
+
         return this.getMessageAsync(record)
             .thenCompose(result -> {
                 if (result == null) {

broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java

@@ -324,6 +324,18 @@ public class PopConsumerServiceTest {
         consumerService.shutdown();
     }
 
+    @Test
+    public void reviveSkipIfGroupAbsent() {
+        String groupName = "PopGroupAbsent";
+        brokerController.getBrokerConfig().setPopReviveSkipIfGroupAbsent(true);
+        PopConsumerRecord record = Mockito.mock(PopConsumerRecord.class);
+        Mockito.when(record.getGroupId()).thenReturn(groupName);
+        Mockito.when(brokerController.getSubscriptionGroupManager()
+            .containsSubscriptionGroup(groupName)).thenReturn(false);
+        CompletableFuture<Boolean> result = consumerService.revive(record);
+        Assert.assertTrue(result.join());
+    }
+
     @Test
     public void reviveRetryTest() {
         Mockito.when(brokerController.getTopicConfigManager().selectTopicConfig(topicId)).thenReturn(null);
@@ -393,6 +405,8 @@ public class PopConsumerServiceTest {
     @Test
     public void reviveBackoffRetryTest() {
         Mockito.when(brokerController.getEscapeBridge()).thenReturn(Mockito.mock(EscapeBridge.class));
+        Mockito.when(brokerController.getSubscriptionGroupManager()
+            .containsSubscriptionGroup(anyString())).thenReturn(true);
         PopConsumerService consumerServiceSpy = Mockito.spy(consumerService);
 
         consumerService.getPopConsumerStore().start();

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -251,6 +251,7 @@ public class BrokerConfig extends BrokerIdentity {
     private int popReviveMaxReturnSizePerRead = 16 * 1024;
     private int popReviveConcurrency = 32;
     private int popReviveMaxAttemptTimes = 16;
+    private boolean popReviveSkipIfGroupAbsent = true;
     // each message queue will have a corresponding retry queue
     private boolean useSeparateRetryQueue = false;
     private boolean realTimeNotifyConsumerChange = true;
@@ -699,6 +700,14 @@ public class BrokerConfig extends BrokerIdentity {
         this.popReviveMaxAttemptTimes = popReviveMaxAttemptTimes;
     }
 
+    public boolean isPopReviveSkipIfGroupAbsent() {
+        return popReviveSkipIfGroupAbsent;
+    }
+
+    public void setPopReviveSkipIfGroupAbsent(boolean popReviveSkipIfGroupAbsent) {
+        this.popReviveSkipIfGroupAbsent = popReviveSkipIfGroupAbsent;
+    }
+
     public boolean isTraceOn() {
         return traceOn;
     }