IoTConsensusV2: Fix npe when data region is not existed (#14180)

apache · Nov 22, 2024 · d82dffd · d82dffd
1 parent 0ec59a3
commit d82dffd
Showing 1 changed file with 24 additions and 10 deletions.
diff --git a/.../java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java b/.../java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
@@ -88,6 +88,8 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
+import static org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils.generateTsFileResource;
+
 public class PipeConsensusReceiver {
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusReceiver.class);
   private static final IoTDBConfig IOTDB_CONFIG = IoTDBDescriptor.getInstance().getConfig();
@@ -653,9 +655,19 @@ private TPipeConsensusTransferResp checkNonFinalFileSeal(
 
   private TSStatus loadFileToDataRegion(String filePath, ProgressIndex progressIndex)
       throws IOException, LoadFileException {
-    StorageEngine.getInstance()
-        .getDataRegion(((DataRegionId) consensusGroupId))
-        .loadNewTsFile(generateTsFileResource(filePath, progressIndex), true, false);
+    DataRegion region =
+        StorageEngine.getInstance().getDataRegion(((DataRegionId) consensusGroupId));
+    if (region != null) {
+      TsFileResource resource = generateTsFileResource(filePath, progressIndex);
+      region.loadNewTsFile(resource, true, false);
+    } else {
+      // Data region is null indicates that dr has been removed or migrated. In those cases, there
+      // is no need to replicate data. we just return success to avoid leader keeping retry
+      LOGGER.info(
+          "PipeConsensus-PipeName-{}: skip load tsfile-{} when sealing, because this region has been removed or migrated.",
+          consensusPipeName,
+          filePath);
+    }
     return RpcUtils.SUCCESS_STATUS;
   }
 
@@ -689,14 +701,16 @@ private void updateWritePointCountMetrics(
   }
 
   private void updateWritePointCountMetrics(long writePointCount) {
-    final DataRegion dataRegion =
-        StorageEngine.getInstance().getDataRegion(((DataRegionId) consensusGroupId));
-    dataRegion
-        .getNonSystemDatabaseName()
+    Optional.ofNullable(
+            StorageEngine.getInstance().getDataRegion(((DataRegionId) consensusGroupId)))
         .ifPresent(
-            databaseName ->
-                LoadTsFileManager.updateWritePointCountMetrics(
-                    dataRegion, databaseName, writePointCount, true));
+            dataRegion ->
+                dataRegion
+                    .getNonSystemDatabaseName()
+                    .ifPresent(
+                        databaseName ->
+                            LoadTsFileManager.updateWritePointCountMetrics(
+                                dataRegion, databaseName, writePointCount, true)));
   }
 
   private TsFileResource generateTsFileResource(String filePath, ProgressIndex progressIndex)