[IOTDB-4092] Protecting Region creation process by adding CreateRegionGroupsProcedure (apache#7006)

Author: CRZbulabula

confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java

@@ -19,6 +19,7 @@
 package org.apache.iotdb.confignode.client.async.datanode;
 
 import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
 import org.apache.iotdb.common.rpc.thrift.TFlushReq;
@@ -50,6 +51,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -231,18 +233,18 @@ public void sendAsyncRequestToDataNode(
   }
 
   /**
-   * Execute CreateRegionsReq asynchronously
+   * Execute CreateRegionGroupsPlan asynchronously
    *
-   * @param createRegionGroupsPlan CreateRegionsReq
    * @param ttlMap Map<StorageGroupName, TTL>
+   * @return Those RegionGroups that failed to create
    */
-  public void createRegions(
+  public Map<TConsensusGroupId, TRegionReplicaSet> createRegionGroups(
       CreateRegionGroupsPlan createRegionGroupsPlan, Map<String, Long> ttlMap) {
     // Because different requests will be sent to the same node when createRegions,
     // so for CreateRegions use Map<index, TDataNodeLocation>
     Map<Integer, TDataNodeLocation> dataNodeLocationMap = new ConcurrentHashMap<>();
     int index = 0;
-    // Count the datanodes to be sent
+    // Count the DataNodes to be sent
     for (List<TRegionReplicaSet> regionReplicaSets :
         createRegionGroupsPlan.getRegionGroupMap().values()) {
       for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
@@ -252,7 +254,7 @@ public void createRegions(
       }
     }
     if (dataNodeLocationMap.isEmpty()) {
-      return;
+      return new HashMap<>();
     }
     for (int retry = 0; retry < retryNum; retry++) {
       index = 0;
@@ -298,7 +300,7 @@ public void createRegions(
                       retry);
                   break;
                 default:
-                  return;
+                  break;
               }
             } else {
               index++;
@@ -317,6 +319,43 @@ public void createRegions(
         break;
       }
     }
+
+    // Filter RegionGroups that weren't created successfully
+    index = 0;
+    Map<TConsensusGroupId, TRegionReplicaSet> failedRegions = new HashMap<>();
+    for (List<TRegionReplicaSet> regionReplicaSets :
+        createRegionGroupsPlan.getRegionGroupMap().values()) {
+      for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
+        for (TDataNodeLocation dataNodeLocation : regionReplicaSet.getDataNodeLocations()) {
+          if (dataNodeLocationMap.containsKey(index)) {
+            failedRegions
+                .computeIfAbsent(
+                    regionReplicaSet.getRegionId(),
+                    empty -> new TRegionReplicaSet().setRegionId(regionReplicaSet.getRegionId()))
+                .addToDataNodeLocations(dataNodeLocation);
+          }
+          index += 1;
+        }
+      }
+    }
+    return failedRegions;
+  }
+
+  private TCreateSchemaRegionReq genCreateSchemaRegionReq(
+      String storageGroup, TRegionReplicaSet regionReplicaSet) {
+    TCreateSchemaRegionReq req = new TCreateSchemaRegionReq();
+    req.setStorageGroup(storageGroup);
+    req.setRegionReplicaSet(regionReplicaSet);
+    return req;
+  }
+
+  private TCreateDataRegionReq genCreateDataRegionReq(
+      String storageGroup, TRegionReplicaSet regionReplicaSet, long TTL) {
+    TCreateDataRegionReq req = new TCreateDataRegionReq();
+    req.setStorageGroup(storageGroup);
+    req.setRegionReplicaSet(regionReplicaSet);
+    req.setTtl(TTL);
+    return req;
   }
 
   /**
@@ -341,23 +380,6 @@ public void broadCastTheLatestConfigNodeGroup(
     }
   }
 
-  private TCreateSchemaRegionReq genCreateSchemaRegionReq(
-      String storageGroup, TRegionReplicaSet regionReplicaSet) {
-    TCreateSchemaRegionReq req = new TCreateSchemaRegionReq();
-    req.setStorageGroup(storageGroup);
-    req.setRegionReplicaSet(regionReplicaSet);
-    return req;
-  }
-
-  private TCreateDataRegionReq genCreateDataRegionReq(
-      String storageGroup, TRegionReplicaSet regionReplicaSet, long TTL) {
-    TCreateDataRegionReq req = new TCreateDataRegionReq();
-    req.setStorageGroup(storageGroup);
-    req.setRegionReplicaSet(regionReplicaSet);
-    req.setTtl(TTL);
-    return req;
-  }
-
   /**
    * Always call this interface when a DataNode is restarted or removed
    *

confignode/src/main/java/org/apache/iotdb/confignode/client/sync/datanode/SyncDataNodeClientPool.java

@@ -134,14 +134,21 @@ private void deleteRegions(
       List<TConsensusGroupId> regionIds,
       Set<TRegionReplicaSet> deletedRegionSet) {
     for (TConsensusGroupId regionId : regionIds) {
-      LOGGER.debug("Delete region {} ", regionId);
+      LOGGER.info("Try to delete RegionReplica: {} on DataNode: {}", regionId, endPoint);
       final TSStatus status =
           sendSyncRequestToDataNodeWithRetry(
               endPoint, regionId, DataNodeRequestType.DELETE_REGIONS);
+
       if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-        LOGGER.info("DELETE Region {} successfully", regionId);
-        deletedRegionSet.removeIf(k -> k.getRegionId().equals(regionId));
+        LOGGER.info("Delete RegionReplica: {} on DataNode: {} successfully", regionId, endPoint);
+      } else {
+        LOGGER.warn(
+            "Failed to delete RegionReplica: {} on DataNode: {}. You might need to delete it manually",
+            regionId,
+            endPoint);
       }
+
+      deletedRegionSet.removeIf(k -> k.getRegionId().equals(regionId));
     }
   }
 

confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -40,7 +40,7 @@
 import org.apache.iotdb.confignode.consensus.request.write.CreateRegionGroupsPlan;
 import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedurePlan;
-import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsPlan;
+import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionGroupsPlan;
 import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupPlan;
 import org.apache.iotdb.confignode.consensus.request.write.DropFunctionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupPlan;
@@ -143,8 +143,8 @@ public static ConfigPhysicalPlan create(ByteBuffer buffer) throws IOException {
         case CreateRegionGroups:
           req = new CreateRegionGroupsPlan();
           break;
-        case DeleteRegions:
-          req = new DeleteRegionsPlan();
+        case DeleteRegionGroups:
+          req = new DeleteRegionGroupsPlan();
           break;
         case GetSchemaPartition:
           req = new GetSchemaPartitionPlan();

confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -32,7 +32,7 @@ public enum ConfigPhysicalPlanType {
   GetStorageGroup,
   CountStorageGroup,
   CreateRegionGroups,
-  DeleteRegions,
+  DeleteRegionGroups,
   GetSchemaPartition,
   CreateSchemaPartition,
   GetOrCreateSchemaPartition,

confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionGroupsPlan.java

@@ -28,21 +28,26 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
-import java.util.TreeMap;
 
 /** Create regions for specific StorageGroups */
 public class CreateRegionGroupsPlan extends ConfigPhysicalPlan {
 
   // Map<StorageGroupName, List<TRegionReplicaSet>>
-  private final Map<String, List<TRegionReplicaSet>> regionGroupMap;
+  protected final Map<String, List<TRegionReplicaSet>> regionGroupMap;
 
   public CreateRegionGroupsPlan() {
     super(ConfigPhysicalPlanType.CreateRegionGroups);
-    this.regionGroupMap = new TreeMap<>();
+    this.regionGroupMap = new HashMap<>();
+  }
+
+  public CreateRegionGroupsPlan(ConfigPhysicalPlanType type) {
+    super(type);
+    this.regionGroupMap = new HashMap<>();
   }
 
   public Map<String, List<TRegionReplicaSet>> getRegionGroupMap() {

…sus/request/write/DeleteRegionsPlan.java …equest/write/DeleteRegionGroupsPlan.javaconfignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsPlan.java renamed to confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionGroupsPlan.java confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsPlan.java renamed to confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionGroupsPlan.java

@@ -18,62 +18,60 @@
  */
 package org.apache.iotdb.confignode.consensus.request.write;
 
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
 import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
 import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
-import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
-public class DeleteRegionsPlan extends ConfigPhysicalPlan {
+public class DeleteRegionGroupsPlan extends CreateRegionGroupsPlan {
 
-  private final Map<String, List<TConsensusGroupId>> deleteRegionMap;
+  boolean needsDeleteInPartitionTable = true;
 
-  public DeleteRegionsPlan() {
-    super(ConfigPhysicalPlanType.DeleteRegions);
-    this.deleteRegionMap = new HashMap<>();
+  public DeleteRegionGroupsPlan() {
+    super(ConfigPhysicalPlanType.DeleteRegionGroups);
   }
 
-  public void addDeleteRegion(String name, TConsensusGroupId consensusGroupId) {
-    deleteRegionMap.computeIfAbsent(name, empty -> new ArrayList<>()).add(consensusGroupId);
+  public boolean isNeedsDeleteInPartitionTable() {
+    return needsDeleteInPartitionTable;
   }
 
-  public Map<String, List<TConsensusGroupId>> getDeleteRegionMap() {
-    return deleteRegionMap;
+  public void setNeedsDeleteInPartitionTable(boolean needsDeleteInPartitionTable) {
+    this.needsDeleteInPartitionTable = needsDeleteInPartitionTable;
   }
 
   @Override
   protected void serializeImpl(DataOutputStream stream) throws IOException {
-    stream.writeInt(ConfigPhysicalPlanType.DeleteRegions.ordinal());
+    stream.writeInt(ConfigPhysicalPlanType.DeleteRegionGroups.ordinal());
 
-    stream.writeInt(deleteRegionMap.size());
-    for (Map.Entry<String, List<TConsensusGroupId>> consensusGroupIdsEntry :
-        deleteRegionMap.entrySet()) {
-      BasicStructureSerDeUtil.write(consensusGroupIdsEntry.getKey(), stream);
-      stream.writeInt(consensusGroupIdsEntry.getValue().size());
-      for (TConsensusGroupId consensusGroupId : consensusGroupIdsEntry.getValue()) {
-        ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+    stream.writeByte(needsDeleteInPartitionTable ? 1 : 0);
+    stream.writeInt(regionGroupMap.size());
+    for (Map.Entry<String, List<TRegionReplicaSet>> entry : regionGroupMap.entrySet()) {
+      BasicStructureSerDeUtil.write(entry.getKey(), stream);
+      stream.writeInt(entry.getValue().size());
+      for (TRegionReplicaSet regionReplicaSet : entry.getValue()) {
+        ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
       }
     }
   }
 
   @Override
   protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+    needsDeleteInPartitionTable = buffer.get() > 0;
     int length = buffer.getInt();
     for (int i = 0; i < length; i++) {
       String name = BasicStructureSerDeUtil.readString(buffer);
-      deleteRegionMap.put(name, new ArrayList<>());
+      regionGroupMap.put(name, new ArrayList<>());
       int regionNum = buffer.getInt();
       for (int j = 0; j < regionNum; j++) {
-        deleteRegionMap.get(name).add(ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer));
+        regionGroupMap.get(name).add(ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(buffer));
       }
     }
   }
@@ -82,12 +80,12 @@ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
   public boolean equals(Object o) {
     if (this == o) return true;
     if (o == null || getClass() != o.getClass()) return false;
-    DeleteRegionsPlan that = (DeleteRegionsPlan) o;
-    return deleteRegionMap.equals(that.deleteRegionMap);
+    DeleteRegionGroupsPlan that = (DeleteRegionGroupsPlan) o;
+    return regionGroupMap.equals(that.regionGroupMap);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(deleteRegionMap);
+    return Objects.hash(regionGroupMap);
   }
 }

confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -616,22 +616,7 @@ public Object getOrCreateDataPartition(
   }
 
   private TSStatus confirmLeader() {
-    TSStatus result = new TSStatus();
-
-    if (getConsensusManager().isLeader()) {
-      return result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
-    } else {
-      result.setCode(TSStatusCode.NEED_REDIRECTION.getStatusCode());
-      result.setMessage(
-          "The current ConfigNode is not leader, please redirect to a new ConfigNode.");
-
-      TConfigNodeLocation leaderLocation = consensusManager.getLeader();
-      if (leaderLocation != null) {
-        result.setRedirectNode(leaderLocation.getInternalEndPoint());
-      }
-
-      return result;
-    }
+    return getConsensusManager().confirmLeader();
   }
 
   @Override

confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java

@@ -20,6 +20,7 @@
 
 import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.consensus.ConsensusGroupId;
 import org.apache.iotdb.commons.consensus.PartitionRegionId;
 import org.apache.iotdb.commons.exception.BadNodeUrlException;
@@ -36,6 +37,7 @@
 import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
 import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
 import org.apache.iotdb.consensus.config.ConsensusConfig;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -202,6 +204,30 @@ public TConfigNodeLocation getLeader() {
     return null;
   }
 
+  /**
+   * Confirm the current ConfigNode's leadership
+   *
+   * @return SUCCESS_STATUS if the current ConfigNode is leader, NEED_REDIRECTION otherwise
+   */
+  public TSStatus confirmLeader() {
+    TSStatus result = new TSStatus();
+
+    if (isLeader()) {
+      return result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    } else {
+      result.setCode(TSStatusCode.NEED_REDIRECTION.getStatusCode());
+      result.setMessage(
+          "The current ConfigNode is not leader, please redirect to a new ConfigNode.");
+
+      TConfigNodeLocation leaderLocation = getLeader();
+      if (leaderLocation != null) {
+        result.setRedirectNode(leaderLocation.getInternalEndPoint());
+      }
+
+      return result;
+    }
+  }
+
   public ConsensusGroupId getConsensusGroupId() {
     return consensusGroupId;
   }