Merge remote-tracking branch 'upstream/develop' into ISSUE-9959

Author: majialoong

NOTICE

@@ -1,5 +1,5 @@
 Apache RocketMQ
-Copyright 2016-2025 The Apache Software Foundation
+Copyright 2016-2026 The Apache Software Foundation
 
 This product includes software developed at
 The Apache Software Foundation (http://www.apache.org/).

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

@@ -501,7 +501,15 @@ public void changeInvisibilityDuration(long popTime, long invisibleTime,
         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 CompletableFuture<Triple<MessageExt, String, Boolean>> getMessageAsync(Po
     }
 
     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 void ackAsyncTest() {
         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 void reviveRetryTest() {
     @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();

broker/src/test/java/org/apache/rocketmq/broker/subscription/RocksdbGroupConfigTransferTest.java

@@ -17,6 +17,15 @@
 
 package org.apache.rocketmq.broker.subscription;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Stream;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.config.v1.RocksDBSubscriptionGroupManager;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -34,15 +43,6 @@
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
 import static org.mockito.Mockito.when;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -78,24 +78,28 @@ public void destroy() {
         if (notToBeExecuted()) {
             return;
         }
-        Path pathToBeDeleted = Paths.get(basePath);
-
-        try {
-            Files.walk(pathToBeDeleted)
-                    .sorted(Comparator.reverseOrder())
-                    .forEach(path -> {
-                        try {
-                            Files.delete(path);
-                        } catch (IOException e) {
-                            // ignore
-                        }
-                    });
-        } catch (IOException e) {
-            // ignore
-        }
+
         if (rocksDBSubscriptionGroupManager != null) {
             rocksDBSubscriptionGroupManager.stop();
         }
+
+        Path root = Paths.get(basePath);
+        if (Files.notExists(root)) {
+            return;
+        }
+
+        try (Stream<Path> walk = Files.walk(root)) {
+            walk.sorted(Comparator.reverseOrder())
+                .forEach(p -> {
+                    try {
+                        Files.deleteIfExists(p);
+                    } catch (IOException e) {
+                        // ignore
+                    }
+                });
+        } catch (IOException e) {
+                // ignore
+        }
     }
 
 

client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java

@@ -286,21 +286,19 @@ private boolean rebalanceByTopic(final String topic, final boolean isOrder) {
             }
             case CLUSTERING: {
                 Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
-                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
-                if (null == mqSet) {
+                if (null == mqSet || mqSet.isEmpty()) {
                     if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                         this.messageQueueChanged(topic, Collections.<MessageQueue>emptySet(), Collections.<MessageQueue>emptySet());
                         log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
                     }
+                    break;
                 }
 
-                if (null == cidAll) {
+                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
+                if (null == cidAll || cidAll.isEmpty()) {
                     log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
-                }
-
-                if (mqSet != null && cidAll != null) {
-                    List<MessageQueue> mqAll = new ArrayList<>();
-                    mqAll.addAll(mqSet);
+                } else {
+                    List<MessageQueue> mqAll = new ArrayList<>(mqSet);
 
                     Collections.sort(mqAll);
                     Collections.sort(cidAll);

client/src/main/java/org/apache/rocketmq/client/latency/MQFaultStrategy.java

@@ -21,8 +21,9 @@
 import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
 import org.apache.rocketmq.client.impl.producer.TopicPublishInfo.QueueFilter;
 import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
 
-public class MQFaultStrategy {
+public class MQFaultStrategy implements StartAndShutdown {
     private LatencyFaultTolerance<String> latencyFaultTolerance;
     private volatile boolean sendLatencyFaultEnable;
     private volatile boolean startDetectorEnable;
@@ -130,6 +131,11 @@ public void startDetector() {
         this.latencyFaultTolerance.startDetector();
     }
 
+    @Override
+    public void start() throws Exception {
+        this.startDetector();
+    }
+
     public void shutdown() {
         this.latencyFaultTolerance.shutdown();
     }

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 void setPopReviveMaxAttemptTimes(int popReviveMaxAttemptTimes) {
         this.popReviveMaxAttemptTimes = popReviveMaxAttemptTimes;
     }
 
+    public boolean isPopReviveSkipIfGroupAbsent() {
+        return popReviveSkipIfGroupAbsent;
+    }
+
+    public void setPopReviveSkipIfGroupAbsent(boolean popReviveSkipIfGroupAbsent) {
+        this.popReviveSkipIfGroupAbsent = popReviveSkipIfGroupAbsent;
+    }
+
     public boolean isTraceOn() {
         return traceOn;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java

@@ -122,7 +122,7 @@ protected static void initConfiguration(CommandLineArgument commandLineArgument)
             System.setProperty(Configuration.CONFIG_PATH_PROPERTY, commandLineArgument.getProxyConfigPath());
         }
         ConfigurationManager.initEnv();
-        ConfigurationManager.intConfig();
+        ConfigurationManager.initConfig();
         setConfigFromCommandLineArgument(commandLineArgument);
         log.info("Current configuration: " + ConfigurationManager.formatProxyConfig());
 

proxy/src/main/java/org/apache/rocketmq/proxy/config/ConfigurationManager.java

@@ -40,7 +40,7 @@ public static void initEnv() {
         }
     }
 
-    public static void intConfig() throws Exception {
+    public static void initConfig() throws Exception {
         configuration = new Configuration();
         configuration.init();
     }

proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java

@@ -1226,10 +1226,6 @@ public void setMetricsExporterType(MetricsExporterType metricsExporterType) {
         this.metricsExporterType = metricsExporterType;
     }
 
-    public void setMetricsExporterType(int metricsExporterType) {
-        this.metricsExporterType = MetricsExporterType.valueOf(metricsExporterType);
-    }
-
     public void setMetricsExporterType(String metricsExporterType) {
         this.metricsExporterType = MetricsExporterType.valueOf(metricsExporterType);
     }