[GitHub] [iotdb] mychaow commented on a change in pull request #3191: New features of cluster scalability and multi-raft

[GitBox <gi...@apache.org>]

mychaow commented on a change in pull request #3191:
URL: https://github.com/apache/iotdb/pull/3191#discussion_r643918392



##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/snapshot/FileSnapshot.java
##########
@@ -195,64 +194,70 @@ public String toString() {
     }
 
     @Override
-    public void install(FileSnapshot snapshot, int slot) throws SnapshotInstallationException {
+    public void install(FileSnapshot snapshot, int slot, boolean isDataMigration)
+        throws SnapshotInstallationException {
       try {
         logger.info("Starting to install a snapshot {} into slot[{}]", snapshot, slot);
         installFileSnapshotSchema(snapshot);
         logger.info("Schemas in snapshot are registered");
-
-        SlotStatus status = slotManager.getStatus(slot);
-        if (status == SlotStatus.PULLING) {
-          // as the schemas are set, writes can proceed
-          slotManager.setToPullingWritable(slot);
-          logger.debug("{}: slot {} is now pulling writable", name, slot);
+        if (isDataMigration) {
+          SlotStatus status = slotManager.getStatus(slot);
+          if (status == SlotStatus.PULLING) {
+            // as the schemas are set, writes can proceed
+            slotManager.setToPullingWritable(slot);
+            logger.debug("{}: slot {} is now pulling writable", name, slot);
+          }
         }
-
-        installFileSnapshotFiles(snapshot, slot);
+        installFileSnapshotFiles(snapshot, slot, isDataMigration);
       } catch (PullFileException e) {
         throw new SnapshotInstallationException(e);
       }
     }
 
     @Override
-    public void install(Map<Integer, FileSnapshot> snapshotMap)
+    public void install(Map<Integer, FileSnapshot> snapshotMap, boolean isDataMigration)
         throws SnapshotInstallationException {
       logger.info("Starting to install snapshots {}", snapshotMap);
-      installSnapshot(snapshotMap);
+      installSnapshot(snapshotMap, isDataMigration);
     }
 
-    private void installSnapshot(Map<Integer, FileSnapshot> snapshotMap)
+    private void installSnapshot(Map<Integer, FileSnapshot> snapshotMap, boolean isDataMigration)
         throws SnapshotInstallationException {
-      // ensure StorageGroups are synchronized
-      try {
-        dataGroupMember.getMetaGroupMember().syncLeaderWithConsistencyCheck(true);
-      } catch (CheckConsistencyException e) {
-        throw new SnapshotInstallationException(e);
-      }
-
-      for (FileSnapshot value : snapshotMap.values()) {
-        installFileSnapshotSchema(value);
-      }
-
+      // In data migration, meta group member other than new node does not need to synchronize the
+      // leader, because data migration must be carried out after meta group applied add/remove node
+      // log.
+      dataGroupMember
+          .getMetaGroupMember()
+          .syncLocalApply(
+              dataGroupMember.getMetaGroupMember().getPartitionTable().getLastMetaLogIndex() - 1,
+              false);
       for (Entry<Integer, FileSnapshot> integerSnapshotEntry : snapshotMap.entrySet()) {
         Integer slot = integerSnapshotEntry.getKey();
-        SlotStatus status = slotManager.getStatus(slot);
-        if (status == SlotStatus.PULLING) {
-          // as schemas are set, writes can proceed
-          slotManager.setToPullingWritable(slot);
-          logger.debug("{}: slot {} is now pulling writable", name, slot);
+        FileSnapshot snapshot = integerSnapshotEntry.getValue();
+        installFileSnapshotSchema(snapshot);
+        if (isDataMigration) {
+          SlotStatus status = slotManager.getStatus(slot);
+          if (status == SlotStatus.PULLING) {
+            // as schemas are set, writes can proceed

Review comment:
       comment is not right

##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
##########
@@ -677,6 +677,9 @@ public boolean matchTerm(long term, long index) {
    */
   void applyEntries(List<Log> entries) {
     for (Log entry : entries) {
+      if (entry.isApplied()) {

Review comment:
       Is this judgment useful? When will apply a log that has been applied（except for restart）?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org