[ISSUE #8576] Support Creating or Updating Subscription Groups in Batch

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

@@ -131,6 +131,7 @@
 import org.apache.rocketmq.remoting.protocol.body.QuerySubscriptionResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
 import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList;
 import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -282,6 +283,8 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx,
                 return this.unlockBatchMQ(ctx, request);
             case RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP:
                 return this.updateAndCreateSubscriptionGroup(ctx, request);
+            case RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST:
+                return this.updateAndCreateSubscriptionGroupList(ctx, request);
             case RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG:
                 return this.getAllSubscriptionGroup(ctx, request);
             case RequestCode.DELETE_SUBSCRIPTIONGROUP:
@@ -1571,6 +1574,41 @@ private RemotingCommand updateAndCreateSubscriptionGroup(ChannelHandlerContext c
         return response;
     }
 
+    private RemotingCommand updateAndCreateSubscriptionGroupList(ChannelHandlerContext ctx, RemotingCommand request) {
+        final long startTime = System.nanoTime();
+
+        final SubscriptionGroupList subscriptionGroupList = SubscriptionGroupList.decode(request.getBody(), SubscriptionGroupList.class);
+        final List<SubscriptionGroupConfig> groupConfigList = subscriptionGroupList.getGroupConfigList();
+
+        final StringBuilder builder = new StringBuilder();
+        for (SubscriptionGroupConfig config : groupConfigList) {
+            builder.append(config.getGroupName()).append(";");
+        }
+        final String groupNames = builder.toString();
+        LOGGER.info("AdminBrokerProcessor#updateAndCreateSubscriptionGroupList: groupNames: {}, called by {}",
+            groupNames,
+            RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+
+        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
+        try {
+            this.brokerController.getSubscriptionGroupManager().updateSubscriptionGroupConfigList(groupConfigList);
+            response.setCode(ResponseCode.SUCCESS);
+            response.setRemark(null);
+        } finally {
+            long executionTime = (System.nanoTime() - startTime) / 1000000L;
+            LOGGER.info("executionTime of create updateAndCreateSubscriptionGroupList: {} is {} ms", groupNames, executionTime);
+            InvocationStatus status = response.getCode() == ResponseCode.SUCCESS ?
+                InvocationStatus.SUCCESS : InvocationStatus.FAILURE;
+            Attributes attributes = BrokerMetricsManager.newAttributesBuilder()
+                .put(LABEL_INVOCATION_STATUS, status.getName())
+                .build();
+            BrokerMetricsManager.consumerGroupCreateExecuteTime.record(executionTime, attributes);
+        }
+
+        return response;
+    }
+
+
     private void initConsumerOffset(String clientHost, String groupName, int mode, TopicConfig topicConfig) {
         String topic = topicConfig.getTopicName();
         for (int queueId = 0; queueId < topicConfig.getReadQueueNums(); queueId++) {

broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
@@ -138,6 +139,11 @@ protected SubscriptionGroupConfig removeSubscriptionGroupConfig(String groupName
     }
 
     public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config) {
+        updateSubscriptionGroupConfigWithoutPersist(config);
+        this.persist();
+    }
+
+    private void updateSubscriptionGroupConfigWithoutPersist(SubscriptionGroupConfig config) {
         Map<String, String> newAttributes = request(config);
         Map<String, String> currentAttributes = current(config.getGroupName());
 
@@ -157,7 +163,13 @@ public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config)
         }
 
         updateDataVersion();
+    }
 
+    public void updateSubscriptionGroupConfigList(List<SubscriptionGroupConfig> configList) {
+        if (null == configList || configList.isEmpty()) {
+            return;
+        }
+        configList.forEach(this::updateSubscriptionGroupConfigWithoutPersist);
         this.persist();
     }
 

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

@@ -18,11 +18,12 @@
 package org.apache.rocketmq.broker.subscription;
 
 import com.google.common.collect.ImmutableMap;
-
 import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
-
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.SubscriptionGroupAttributes;
@@ -39,7 +40,11 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
 
 @RunWith(MockitoJUnitRunner.class)
 public class SubscriptionGroupManagerTest {
@@ -113,4 +118,52 @@ public void updateSubscriptionGroupConfig() {
     private boolean notToBeExecuted() {
         return MixAll.isMac();
     }
+    @Test
+    public void testUpdateSubscriptionGroupConfigList_NullConfigList() {
+        if (notToBeExecuted()) {
+            return;
+        }
+
+        subscriptionGroupManager.updateSubscriptionGroupConfigList(null);
+        // Verifying that persist() is not called
+        verify(subscriptionGroupManager, never()).persist();
+    }
+
+    @Test
+    public void testUpdateSubscriptionGroupConfigList_EmptyConfigList() {
+        if (notToBeExecuted()) {
+            return;
+        }
+
+        subscriptionGroupManager.updateSubscriptionGroupConfigList(Collections.emptyList());
+        // Verifying that persist() is not called
+        verify(subscriptionGroupManager, never()).persist();
+    }
+
+    @Test
+    public void testUpdateSubscriptionGroupConfigList_ValidConfigList() {
+        if (notToBeExecuted()) {
+            return;
+        }
+
+        final List<SubscriptionGroupConfig> configList = new LinkedList<>();
+        final List<String> groupNames = new LinkedList<>();
+        for (int i = 0; i < 10; i++) {
+            SubscriptionGroupConfig config = new SubscriptionGroupConfig();
+            String groupName = String.format("group-%d", i);
+            config.setGroupName(groupName);
+            configList.add(config);
+            groupNames.add(groupName);
+        }
+
+        subscriptionGroupManager.updateSubscriptionGroupConfigList(configList);
+
+        // Verifying that persist() is called once
+        verify(subscriptionGroupManager, times(1)).persist();
+
+        groupNames.forEach(groupName ->
+            assertThat(subscriptionGroupManager.getSubscriptionGroupTable().get(groupName)).isNotNull());
+
+    }
+
 }

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -137,6 +137,7 @@
 import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
 import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
 import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList;
 import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -400,6 +401,22 @@ public void createSubscriptionGroup(final String addr, final SubscriptionGroupCo
 
     }
 
+    public void createSubscriptionGroupList(final String address, final List<SubscriptionGroupConfig> configs,
+        final long timeoutMillis) throws RemotingException, InterruptedException, MQClientException {
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST, null);
+        SubscriptionGroupList requestBody = new SubscriptionGroupList(configs);
+        request.setBody(requestBody.encode());
+
+        RemotingCommand response = this.remotingClient.invokeSync(
+            MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), address), request, timeoutMillis);
+        assert response != null;
+        if (response.getCode() == ResponseCode.SUCCESS) {
+            return;
+        }
+
+        throw new MQClientException(response.getCode(), response.getRemark());
+    }
+
     public void createTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig,
         final long timeoutMillis)
         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java

@@ -148,6 +148,8 @@ public class RequestCode {
 
     public static final int GET_TOPICS_BY_CLUSTER = 224;
 
+    public static final int UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST = 225;
+
     public static final int QUERY_TOPICS_BY_CONSUMER = 343;
     public static final int QUERY_SUBSCRIPTION_BY_CONSUMER = 345;
 

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/SubscriptionGroupList.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.remoting.protocol.body;
+
+import java.util.List;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+
+public class SubscriptionGroupList extends RemotingSerializable {
+    @CFNotNull
+    private List<SubscriptionGroupConfig> groupConfigList;
+
+    public SubscriptionGroupList() {}
+
+    public SubscriptionGroupList(List<SubscriptionGroupConfig> groupConfigList) {
+        this.groupConfigList = groupConfigList;
+    }
+
+    public List<SubscriptionGroupConfig> getGroupConfigList() {
+        return groupConfigList;
+    }
+
+    public void setGroupConfigList(List<SubscriptionGroupConfig> groupConfigList) {
+        this.groupConfigList = groupConfigList;
+    }
+
+}

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java

@@ -238,6 +238,12 @@ public void createAndUpdateSubscriptionGroupConfig(String addr,
         defaultMQAdminExtImpl.createAndUpdateSubscriptionGroupConfig(addr, config);
     }
 
+    @Override
+    public void createAndUpdateSubscriptionGroupConfigList(String brokerAddr,
+        List<SubscriptionGroupConfig> configs) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExtImpl.createAndUpdateSubscriptionGroupConfigList(brokerAddr, configs);
+    }
+
     @Override
     public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group)
         throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -318,6 +318,12 @@ public void createAndUpdateSubscriptionGroupConfig(String addr,
         this.mqClientInstance.getMQClientAPIImpl().createSubscriptionGroup(addr, config, timeoutMillis);
     }
 
+    @Override
+    public void createAndUpdateSubscriptionGroupConfigList(String brokerAddr,
+        List<SubscriptionGroupConfig> configs) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        this.mqClientInstance.getMQClientAPIImpl().createSubscriptionGroupList(brokerAddr, configs, timeoutMillis);
+    }
+
     @Override
     public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr,
         String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java

@@ -118,6 +118,10 @@ void createAndUpdateSubscriptionGroupConfig(final String addr,
         final SubscriptionGroupConfig config) throws RemotingException,
         MQBrokerException, InterruptedException, MQClientException;
 
+    void createAndUpdateSubscriptionGroupConfigList(String brokerAddr,
+        List<SubscriptionGroupConfig> configs) throws RemotingException,
+        MQBrokerException, InterruptedException, MQClientException;
+
     SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr,
         final String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException;
 

tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java

@@ -67,6 +67,7 @@
 import org.apache.rocketmq.tools.command.consumer.GetConsumerConfigSubCommand;
 import org.apache.rocketmq.tools.command.consumer.SetConsumeModeSubCommand;
 import org.apache.rocketmq.tools.command.consumer.StartMonitoringSubCommand;
+import org.apache.rocketmq.tools.command.consumer.UpdateSubGroupListSubCommand;
 import org.apache.rocketmq.tools.command.consumer.UpdateSubGroupSubCommand;
 import org.apache.rocketmq.tools.command.container.AddBrokerSubCommand;
 import org.apache.rocketmq.tools.command.container.RemoveBrokerSubCommand;
@@ -192,6 +193,7 @@ public static void initCommand() {
         initCommand(new UpdateTopicListSubCommand());
         initCommand(new DeleteTopicSubCommand());
         initCommand(new UpdateSubGroupSubCommand());
+        initCommand(new UpdateSubGroupListSubCommand());
         initCommand(new SetConsumeModeSubCommand());
         initCommand(new DeleteSubscriptionGroupCommand());
         initCommand(new UpdateBrokerConfigSubCommand());