Fix the problem that writes duplicate TConsensusGroupId when repairing data partition table (#17416)

Author: zerolbsony

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/partition/DataPartitionTableIntegrityCheckProcedure.java

@@ -332,6 +332,7 @@ private Flow analyzeMissingPartitions(final ConfigNodeProcedureEnv env) {
 
       Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TConsensusGroupId>>>
           seriesPartitionMap = localDataPartitionTable.get(database);
+      long localEarliestSlotStartTime = Long.MAX_VALUE;
       for (Map.Entry<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TConsensusGroupId>>>
           seriesPartitionEntry : seriesPartitionMap.entrySet()) {
         Map<TTimePartitionSlot, List<TConsensusGroupId>> tTimePartitionSlotListMap =
@@ -346,14 +347,17 @@ private Flow analyzeMissingPartitions(final ConfigNodeProcedureEnv env) {
                 .min(Comparator.comparingLong(TTimePartitionSlot::getStartTime))
                 .orElse(null);
 
-        if (localEarliestSlot.getStartTime()
-            > TimePartitionUtils.getStartTimeByPartitionId(earliestTimeslot)) {
-          databasesWithLostDataPartition.add(database);
-          LOG.warn(
-              "[DataPartitionIntegrity] Database {} has lost timeslot {} in its data table partition, and this issue needs to be repaired",
-              database,
-              earliestTimeslot);
-        }
+        localEarliestSlotStartTime =
+            Math.min(localEarliestSlotStartTime, localEarliestSlot.getStartTime());
+      }
+
+      if (localEarliestSlotStartTime
+          > TimePartitionUtils.getStartTimeByPartitionId(earliestTimeslot)) {
+        databasesWithLostDataPartition.add(database);
+        LOG.warn(
+            "[DataPartitionIntegrity] Database {} has lost timeslot {} in its data table partition, and this issue needs to be repaired",
+            database,
+            earliestTimeslot);
       }
     }
 

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java

@@ -37,10 +37,12 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.Vector;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListMap;
@@ -280,7 +282,16 @@ public void merge(SeriesPartitionTable sourceMap) {
     if (sourceMap == null) return;
     sourceMap.seriesPartitionMap.forEach(
         (timeSlot, groups) -> {
-          this.seriesPartitionMap.computeIfAbsent(timeSlot, k -> new ArrayList<>()).addAll(groups);
+          List<TConsensusGroupId> groupList =
+              this.seriesPartitionMap.computeIfAbsent(timeSlot, k -> new ArrayList<>());
+          synchronized (groupList) {
+            Set<TConsensusGroupId> groupSet = new HashSet<>(groupList);
+            for (TConsensusGroupId groupId : groups) {
+              if (!groupSet.contains(groupId)) {
+                groupList.add(groupId);
+              }
+            }
+          }
         });
   }