feat: add CLI to update topic

cli/src/main/java/com/automq/rocketmq/cli/ConsoleHelper.java

@@ -0,0 +1,68 @@
+/*
+ * 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 com.automq.rocketmq.cli;
+
+import apache.rocketmq.controller.v1.MessageQueueAssignment;
+import apache.rocketmq.controller.v1.OngoingMessageQueueReassignment;
+import apache.rocketmq.controller.v1.Topic;
+import de.vandermeer.asciitable.AT_Row;
+import de.vandermeer.asciitable.AsciiTable;
+import de.vandermeer.skb.interfaces.transformers.textformat.TextAlignment;
+import java.util.List;
+
+public class ConsoleHelper {
+    public static void printTable(Topic topic) {
+        AsciiTable topicTable = new AsciiTable();
+        topicTable.addRule();
+        topicTable.addRow("TOPIC ID", "TOPIC NAME");
+        topicTable.addRule();
+        topicTable.addRow(topic.getTopicId(), topic.getName());
+        topicTable.addRule();
+        String render = topicTable.render();
+        System.out.println(render);
+
+        AsciiTable assignmentTable = new AsciiTable();
+        assignmentTable.addRule();
+        AT_Row row = assignmentTable.addRow(null, "ASSIGNMENT");
+        row.getCells().get(1).getContext().setTextAlignment(TextAlignment.CENTER);
+        assignmentTable.addRule();
+        assignmentTable.addRow("NODE ID", "QUEUE ID");
+        assignmentTable.addRule();
+        for (MessageQueueAssignment assignment : topic.getAssignmentsList()) {
+            assignmentTable.addRow(assignment.getNodeId(), assignment.getQueue().getQueueId());
+            assignmentTable.addRule();
+        }
+        render = assignmentTable.render();
+        System.out.println(render);
+
+        List<OngoingMessageQueueReassignment> ongoing = topic.getReassignmentsList();
+        if (!ongoing.isEmpty()) {
+            AsciiTable reassignmentTable = new AsciiTable();
+            assignmentTable.addRule();
+            row = assignmentTable.addRow(null, "ON-GOING REASSIGNMENT");
+            row.getCells().get(1).getContext().setTextAlignment(TextAlignment.CENTER);
+            reassignmentTable.addRule();
+            reassignmentTable.addRow("SRC NODE ID", "DST NODE ID", "QUEUE ID");
+            reassignmentTable.addRule();
+            for (OngoingMessageQueueReassignment reassignment : ongoing) {
+                reassignmentTable.addRow(reassignment.getSrcNodeId(), reassignment.getDstNodeId(), reassignment.getQueue().getQueueId());
+                reassignmentTable.addRule();
+            }
+        }
+    }
+}

cli/src/main/java/com/automq/rocketmq/cli/DescribeTopic.java

@@ -17,14 +17,8 @@
 
 package com.automq.rocketmq.cli;
 
-import apache.rocketmq.controller.v1.MessageQueueAssignment;
-import apache.rocketmq.controller.v1.OngoingMessageQueueReassignment;
 import apache.rocketmq.controller.v1.Topic;
 import com.automq.rocketmq.controller.metadata.GrpcControllerClient;
-import de.vandermeer.asciitable.AT_Row;
-import de.vandermeer.asciitable.AsciiTable;
-import de.vandermeer.skb.interfaces.transformers.textformat.TextAlignment;
-import java.util.List;
 import java.util.concurrent.Callable;
 import picocli.CommandLine;
 
@@ -46,44 +40,7 @@ public Void call() throws Exception {
                 System.err.printf("Topic '%s' is not found%n%n", topicName);
                 return null;
             }
-            AsciiTable topicTable = new AsciiTable();
-            topicTable.addRule();
-            topicTable.addRow("TOPIC ID", "TOPIC NAME");
-            topicTable.addRule();
-            topicTable.addRow(topic.getTopicId(), topic.getName());
-            topicTable.addRule();
-            String render = topicTable.render();
-            System.out.println(render);
-
-            AsciiTable assignmentTable = new AsciiTable();
-            assignmentTable.addRule();
-            AT_Row row = assignmentTable.addRow(null, "ASSIGNMENT");
-            row.getCells().get(1).getContext().setTextAlignment(TextAlignment.CENTER);
-            assignmentTable.addRule();
-            assignmentTable.addRow("NODE ID", "QUEUE ID");
-            assignmentTable.addRule();
-            for (MessageQueueAssignment assignment : topic.getAssignmentsList()) {
-                assignmentTable.addRow(assignment.getNodeId(), assignment.getQueue().getQueueId());
-                assignmentTable.addRule();
-            }
-            render = assignmentTable.render();
-            System.out.println(render);
-
-            List<OngoingMessageQueueReassignment> ongoing = topic.getReassignmentsList();
-            if (!ongoing.isEmpty()) {
-                AsciiTable reassignmentTable = new AsciiTable();
-                assignmentTable.addRule();
-                row = assignmentTable.addRow(null, "ON-GOING REASSIGNMENT");
-                row.getCells().get(1).getContext().setTextAlignment(TextAlignment.CENTER);
-                reassignmentTable.addRule();
-                reassignmentTable.addRow("SRC NODE ID", "DST NODE ID", "QUEUE ID");
-                reassignmentTable.addRule();
-                for (OngoingMessageQueueReassignment reassignment : ongoing) {
-                    reassignmentTable.addRow(reassignment.getSrcNodeId(), reassignment.getDstNodeId(), reassignment.getQueue().getQueueId());
-                    reassignmentTable.addRule();
-                }
-            }
-
+            ConsoleHelper.printTable(topic);
         }
         return null;
     }

cli/src/main/java/com/automq/rocketmq/cli/MQAdmin.java

@@ -27,6 +27,7 @@
     subcommands = {
         CreateTopic.class,
         DescribeTopic.class,
+        UpdateTopic.class,
         CreateGroup.class,
         DescribeGroup.class,
         ProduceMessage.class,

cli/src/main/java/com/automq/rocketmq/cli/UpdateTopic.java

@@ -0,0 +1,67 @@
+/*
+ * 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 com.automq.rocketmq.cli;
+
+import apache.rocketmq.controller.v1.Topic;
+import apache.rocketmq.controller.v1.UpdateTopicRequest;
+import com.automq.rocketmq.controller.metadata.GrpcControllerClient;
+import com.google.common.base.Strings;
+import java.util.concurrent.Callable;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "updateTopic", mixinStandardHelpOptions = true, showDefaultValues = true)
+public class UpdateTopic implements Callable<Void> {
+
+    @CommandLine.Option(names = {"-i", "--topicId"}, description = "Topic ID", required = true)
+    long topicId;
+
+    @CommandLine.Option(names = {"-t", "--topicName"}, description = "Topic name")
+    String topicName;
+
+    @CommandLine.Option(names = {"-q", "--queueNumber"}, description = "Queue number")
+    int queueNumber = 0;
+
+    @CommandLine.ParentCommand
+    MQAdmin mqAdmin;
+
+    @Override
+    public Void call() throws Exception {
+        try (GrpcControllerClient client = new GrpcControllerClient()) {
+            Topic topic = client.describeTopic(mqAdmin.endpoint, topicId, null)
+                .join();
+            if (null == topic) {
+                System.err.printf("Topic '%s' is not found%n%n", topicName);
+                return null;
+            }
+
+            UpdateTopicRequest.Builder builder = UpdateTopicRequest.newBuilder()
+                .setTopicId(topicId);
+            if (queueNumber > topic.getCount()) {
+                builder.setCount(queueNumber);
+            }
+
+            if (!Strings.isNullOrEmpty(topicName)) {
+                builder.setName(topicName);
+            }
+            client.updateTopic(mqAdmin.endpoint, builder.build()).join();
+            topic = client.describeTopic(mqAdmin.endpoint, topicId, null).join();
+            ConsoleHelper.printTable(topic);
+        }
+        return null;
+    }
+}

controller/src/main/java/com/automq/rocketmq/controller/ControllerServiceImpl.java

@@ -263,6 +263,7 @@ public void updateTopic(UpdateTopicRequest request, StreamObserver<UpdateTopicRe
             if (request.getCount() > 0) {
                 queueNumber = request.getCount();
             }
+
             this.metadataStore.updateTopic(request.getTopicId(), request.getName(), queueNumber,
                     request.getAcceptMessageTypesList())
                 .whenComplete((res, e) -> {

controller/src/main/java/com/automq/rocketmq/controller/metadata/GrpcControllerClient.java

@@ -238,17 +238,20 @@ public void onFailure(@Nonnull Throwable t) {
     }
 
     @Override
-    public CompletableFuture<Topic> describeTopic(String target, Long topicId,
-        String topicName) {
+    public CompletableFuture<Topic> describeTopic(String target, Long topicId, String topicName) {
 
-        DescribeTopicRequest request = DescribeTopicRequest.newBuilder()
-            .setTopicId(null == topicId ? -1 : topicId)
-            .setTopicName(topicName)
-            .build();
+        DescribeTopicRequest.Builder builder = DescribeTopicRequest.newBuilder();
+        if (null != topicId && topicId > 0) {
+            builder.setTopicId(topicId);
+        }
+
+        if (!Strings.isNullOrEmpty(topicName)) {
+            builder.setTopicName(topicName.trim());
+        }
 
         CompletableFuture<Topic> future = new CompletableFuture<>();
         try {
-            Futures.addCallback(buildStubForTarget(target).describeTopic(request), new FutureCallback<>() {
+            Futures.addCallback(buildStubForTarget(target).describeTopic(builder.build()), new FutureCallback<>() {
                 @Override
                 public void onSuccess(DescribeTopicReply result) {
                     if (result.getStatus().getCode() == Code.OK) {

controller/src/main/java/com/automq/rocketmq/controller/metadata/database/TopicManager.java

@@ -179,12 +179,18 @@ public CompletableFuture<apache.rocketmq.controller.v1.Topic> updateTopic(long t
                         changed = true;
                     }
 
-                    // Update queue nums
+                    // Update queue number
                     if (null != queueNumber && !queueNumber.equals(topic.getQueueNum())) {
                         if (queueNumber > topic.getQueueNum()) {
-                            createQueues(IntStream.range(topic.getQueueNum(), queueNumber), topic.getId(), session);
+                            IntStream range = IntStream.range(topic.getQueueNum(), queueNumber);
+                            List<QueueAssignment> assignments = createQueues(range, topic.getId(), session);
+                            assignmentCache.apply(assignments);
+                            topic.setQueueNum(queueNumber);
+                            changed = true;
+                        } else {
+                            // Ignore queue number field if not enlarged
+                            topic.setQueueNum(null);
                         }
-                        changed = true;
                     }
 
                     if (changed) {

controller/src/main/resources/database/mapper/GroupMapper.xml

@@ -50,26 +50,27 @@
     <select id="byCriteria" resultType="Group" parameterType="GroupCriteria">
         SELECT id, name, status, dead_letter_topic_id, max_delivery_attempt, group_type, create_time, update_time
         FROM consumer_group
-        WHERE 1 = 1
-        <if test="null != id">
-            AND id = #{id}
-        </if>
+        <where>
+            <if test="null != id">
+                AND id = #{id}
+            </if>
 
-        <if test="null != name">
-            AND name = #{name}
-        </if>
+            <if test="null != name">
+                AND name = #{name}
+            </if>
 
-        <if test="null != status">
-            AND status = #{status}
-        </if>
+            <if test="null != status">
+                AND status = #{status}
+            </if>
 
-        <if test="null != topicId">
-            AND dead_letter_topic_id = #{topicId}
-        </if>
+            <if test="null != topicId">
+                AND dead_letter_topic_id = #{topicId}
+            </if>
 
-        <if test="null != updateTime">
-            AND update_time >= #{updateTime}
-        </if>
+            <if test="null != updateTime">
+                AND update_time >= #{updateTime}
+            </if>
+        </where>
     </select>
 
 </mapper>

controller/src/main/resources/database/mapper/TopicMapper.xml

@@ -41,26 +41,20 @@
     <update id="update">
         UPDATE topic
         <set>
-            <if test="null != status">
-                status = #{status},
-            </if>
-            <if test="null != queueNum">
-                queue_num = #{queueNum},
-            </if>
-            <if test="null != acceptMessageTypes">
-                accept_message_types = #{acceptMessageTypes},
-            </if>
+            <if test="null != status">status = #{status},</if>
+            <if test="null != name">name = #{name},</if>
+            <if test="null != queueNum">queue_num = #{queueNum},</if>
+            <if test="null != acceptMessageTypes">accept_message_types = #{acceptMessageTypes},</if>
         </set>
         WHERE id = #{id}
     </update>
 
     <delete id="delete">
         DELETE
         FROM topic
-        WHERE 1 = 1
-        <if test="null != id">
-            AND id = #{id}
-        </if>
+        <where>
+            <if test="null != id">id = #{id}</if>
+        </where>
     </delete>
 
     <delete id="recycle">
@@ -72,25 +66,18 @@
     <select id="get" resultType="Topic">
         SELECT id, name, queue_num, status, create_time, update_time, accept_message_types
         FROM topic
-        WHERE 1 = 1
-        <if test="null != id">
-            AND id = #{id}
-        </if>
-        <if test="null != name and '' != name">
-           AND name = #{name}
-        </if>
+        <where>
+            <if test="null != id">AND id = #{id}</if>
+            <if test="null != name and '' != name">AND name = #{name}</if>
+        </where>
     </select>
 
     <select id="list" resultType="Topic">
         SELECT id, name, queue_num, status, create_time, update_time, accept_message_types
         FROM topic
-        WHERE 1 = 1
-        <if test="status != null">
-            AND status = #{status}
-        </if>
-
-        <if test="updateTime != null">
-            AND update_time >= #{updateTime}
-        </if>
+        <where>
+            <if test="status != null">AND status = #{status}</if>
+            <if test="updateTime != null">AND update_time >= #{updateTime}</if>
+        </where>
     </select>
 </mapper>