From 45946d475697f0b78e4d3b31105f75bb76e7ee19 Mon Sep 17 00:00:00 2001
From: dongeforever <dongeforever@apache.org>
Date: Thu, 18 Nov 2021 10:15:44 +0800
Subject: [PATCH] Ignore the existed queueId in static topi creation

---
 .../topic/UpdateStaticTopicSubCommand.java    | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
index 5cee8c8f..29a7261b 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
@@ -17,6 +17,7 @@
 package org.apache.rocketmq.tools.command.topic;
 
 import com.google.common.collect.ImmutableList;
+import com.sun.xml.internal.ws.api.BindingIDFactory;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionGroup;
@@ -201,22 +202,28 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
 
             //construct the topic configAndMapping
             long epoch = Math.max(maxEpochAndNum.getKey() + 1000, System.currentTimeMillis());
-            newIdToBroker.forEach( (queueId, broker) -> {
+            for (Map.Entry<Integer, String> e : newIdToBroker.entrySet()) {
+                Integer queueId = e.getKey();
+                String value = e.getValue();
+                if (globalIdMap.containsKey(queueId)) {
+                    //ignore the exited
+                    continue;
+                }
                 TopicConfigAndQueueMapping configMapping;
-                if (!existedTopicConfigMap.containsKey(broker)) {
+                if (!existedTopicConfigMap.containsKey(value)) {
                     TopicConfig topicConfig = new TopicConfig(topic, 1, 1);
-                    TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail(topic, queueNum, broker, epoch);
+                    TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail(topic, queueNum, value, epoch);
                     configMapping = new TopicConfigAndQueueMapping(topicConfig, mappingDetail);
                 } else {
-                    configMapping = existedTopicConfigMap.get(broker);
+                    configMapping = existedTopicConfigMap.get(value);
                     configMapping.setWriteQueueNums(configMapping.getWriteQueueNums() + 1);
                     configMapping.setWriteQueueNums(configMapping.getWriteQueueNums() + 1);
                     configMapping.getMappingDetail().setEpoch(epoch);
                     configMapping.getMappingDetail().setTotalQueues(queueNum);
                 }
-                LogicQueueMappingItem mappingItem = new LogicQueueMappingItem(0, configMapping.getWriteQueueNums() - 1, broker, 0, 0, -1, -1, -1);
+                LogicQueueMappingItem mappingItem = new LogicQueueMappingItem(0, configMapping.getWriteQueueNums() - 1, value, 0, 0, -1, -1, -1);
                 configMapping.getMappingDetail().putMappingInfo(queueId, ImmutableList.of(mappingItem));
-            });
+            }
 
             //If some succeed, and others fail, it will cause inconsistent data
             for (Map.Entry<String, TopicConfigAndQueueMapping> entry : existedTopicConfigMap.entrySet()) {
-- 
GitLab