Load: detect region replica set changes due to Region Migration (#14104)

Author: DanielWang2035

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -1176,6 +1176,8 @@ public class IoTDBConfig {
 
   private long loadCleanupTaskExecutionDelayTimeSeconds = 1800L; // 30 min
 
+  private int loadTsFileRetryCountOnRegionChange = 10;
+
   private double loadWriteThroughputBytesPerSecond = -1; // Bytes/s
 
   private boolean loadActiveListeningEnable = true;
@@ -4138,6 +4140,14 @@ public void setLoadCleanupTaskExecutionDelayTimeSeconds(
     this.loadCleanupTaskExecutionDelayTimeSeconds = loadCleanupTaskExecutionDelayTimeSeconds;
   }
 
+  public int getLoadTsFileRetryCountOnRegionChange() {
+    return loadTsFileRetryCountOnRegionChange;
+  }
+
+  public void setLoadTsFileRetryCountOnRegionChange(int loadTsFileRetryCountOnRegionChange) {
+    this.loadTsFileRetryCountOnRegionChange = loadTsFileRetryCountOnRegionChange;
+  }
+
   public double getLoadWriteThroughputBytesPerSecond() {
     return loadWriteThroughputBytesPerSecond;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -2412,6 +2412,11 @@ private void loadLoadTsFileProps(TrimProperties properties) {
             properties.getProperty(
                 "load_clean_up_task_execution_delay_time_seconds",
                 String.valueOf(conf.getLoadCleanupTaskExecutionDelayTimeSeconds()))));
+    conf.setLoadTsFileRetryCountOnRegionChange(
+        Integer.parseInt(
+            properties.getProperty(
+                "load_tsfile_retry_count_on_region_change",
+                String.valueOf(conf.getLoadTsFileRetryCountOnRegionChange()))));
     conf.setLoadWriteThroughputBytesPerSecond(
         Double.parseDouble(
             properties.getProperty(

…db/exception/LoadEmptyFileException.java …ception/load/LoadEmptyFileException.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadEmptyFileException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadEmptyFileException.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadEmptyFileException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadEmptyFileException.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.exception;
+package org.apache.iotdb.db.exception.load;
 
 public class LoadEmptyFileException extends LoadFileException {
 

…otdb/db/exception/LoadFileException.java …db/exception/load/LoadFileException.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadFileException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadFileException.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadFileException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadFileException.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.exception;
+package org.apache.iotdb.db.exception.load;
 
 import org.apache.iotdb.commons.exception.IoTDBException;
 import org.apache.iotdb.rpc.TSStatusCode;

…/db/exception/LoadReadOnlyException.java …xception/load/LoadReadOnlyException.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadReadOnlyException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadReadOnlyException.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadReadOnlyException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadReadOnlyException.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.exception;
+package org.apache.iotdb.db.exception.load;
 
 import org.apache.iotdb.commons.exception.IoTDBException;
 import org.apache.iotdb.rpc.TSStatusCode;

…ion/LoadRuntimeOutOfMemoryException.java …oad/LoadRuntimeOutOfMemoryException.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadRuntimeOutOfMemoryException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadRuntimeOutOfMemoryException.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadRuntimeOutOfMemoryException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadRuntimeOutOfMemoryException.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.exception;
+package org.apache.iotdb.db.exception.load;
 
 public class LoadRuntimeOutOfMemoryException extends RuntimeException {
   public LoadRuntimeOutOfMemoryException(String message) {

…ception/PartitionViolationException.java …on/load/PartitionViolationException.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/PartitionViolationException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/PartitionViolationException.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/PartitionViolationException.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/PartitionViolationException.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.exception;
+package org.apache.iotdb.db.exception.load;
 
 public class PartitionViolationException extends LoadFileException {
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/RegionReplicaSetChangedException.java

@@ -0,0 +1,36 @@
+/*
+ * 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 org.apache.iotdb.db.exception.load;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+
+public class RegionReplicaSetChangedException extends LoadFileException {
+
+  public RegionReplicaSetChangedException(TRegionReplicaSet original, TRegionReplicaSet current) {
+    super(
+        String.format(
+            "Region replica set changed from %s to %s during loading TsFile, maybe due to region migration",
+            original, current));
+  }
+
+  public RegionReplicaSetChangedException() {
+    super("Region replica set changed during loading TsFile, maybe due to region migration");
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java

@@ -22,7 +22,7 @@
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.db.auth.AuthorityChecker;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.exception.LoadFileException;
+import org.apache.iotdb.db.exception.load.LoadFileException;
 import org.apache.iotdb.db.protocol.session.SessionManager;
 import org.apache.iotdb.db.queryengine.common.SessionInfo;
 import org.apache.iotdb.db.queryengine.plan.Coordinator;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java

@@ -23,7 +23,7 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.db.auth.AuthorityChecker;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.exception.LoadFileException;
+import org.apache.iotdb.db.exception.load.LoadFileException;
 import org.apache.iotdb.db.protocol.session.SessionManager;
 import org.apache.iotdb.db.queryengine.common.SessionInfo;
 import org.apache.iotdb.db.queryengine.plan.Coordinator;