[IoTConsensusV2] Fix NPE when transfer tsfile mods #14840

Author: Pengzna

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java

@@ -105,7 +105,7 @@ public class PipeConsensusReceiver {
   private final List<String> receiverBaseDirsName;
   // Used to buffer TsFile when transfer TsFile asynchronously.
   private final PipeConsensusTsFileWriterPool pipeConsensusTsFileWriterPool;
-  private final AtomicReference<File> receiverFileDirWithIdSuffix = new AtomicReference<>();
+  private final AtomicReference<File> receiverFileBaseDir = new AtomicReference<>();
   private final PipeConsensusReceiverMetrics pipeConsensusReceiverMetrics;
   private final FolderManager folderManager;
 
@@ -137,8 +137,7 @@ public PipeConsensusReceiver(
     try {
       initiateTsFileBufferFolder();
       this.pipeConsensusTsFileWriterPool =
-          new PipeConsensusTsFileWriterPool(
-              consensusPipeName, receiverFileDirWithIdSuffix.get().getPath());
+          new PipeConsensusTsFileWriterPool(consensusPipeName, receiverFileBaseDir.get().getPath());
     } catch (Exception e) {
       LOGGER.error("Fail to initiate file buffer folder, Error msg: {}", e.getMessage());
       throw new RuntimeException(e);
@@ -509,9 +508,11 @@ private TPipeConsensusTransferResp handleTransferFileSealWithMods(
     File writingFile = tsFileWriter.getWritingFile();
     RandomAccessFile writingFileWriter = tsFileWriter.getWritingFileWriter();
 
+    File currentWritingDirPath = new File(tsFileWriter.getLocalWritingDirPath());
+
     final List<File> files =
         req.getFileNames().stream()
-            .map(fileName -> new File(receiverFileDirWithIdSuffix.get(), fileName))
+            .map(fileName -> new File(currentWritingDirPath, fileName))
             .collect(Collectors.toList());
     try {
       if (isWritingFileNonAvailable(tsFileWriter)) {
@@ -607,7 +608,7 @@ private TPipeConsensusTransferResp handleTransferFileSealWithMods(
       closeCurrentWritingFileWriter(tsFileWriter, false);
       // Clear the directory instead of only deleting the referenced files in seal request
       // to avoid previously undeleted file being redundant when transferring multi files
-      IoTDBReceiverAgent.cleanPipeReceiverDir(receiverFileDirWithIdSuffix.get());
+      IoTDBReceiverAgent.cleanPipeReceiverDir(currentWritingDirPath);
     }
   }
 
@@ -812,13 +813,13 @@ private boolean isWritingFileOffsetNonCorrect(
   }
 
   private void closeCurrentWritingFileWriter(
-      PipeConsensusTsFileWriter tsFileWriter, boolean fsyncAfterClose) {
+      PipeConsensusTsFileWriter tsFileWriter, boolean fsyncBeforeClose) {
     if (tsFileWriter.getWritingFileWriter() != null) {
       try {
-        tsFileWriter.getWritingFileWriter().close();
-        if (fsyncAfterClose) {
+        if (fsyncBeforeClose) {
           tsFileWriter.getWritingFileWriter().getFD().sync();
         }
+        tsFileWriter.getWritingFileWriter().close();
         LOGGER.info(
             "PipeConsensus-PipeName-{}: Current writing file writer {} was closed.",
             consensusPipeName,
@@ -909,17 +910,17 @@ private void updateWritingFileIfNeeded(
     // Make sure receiver file dir exists
     // This may be useless, because receiver file dir is created when receiver is initiated. just in
     // case.
-    if (!receiverFileDirWithIdSuffix.get().exists()) {
-      if (receiverFileDirWithIdSuffix.get().mkdirs()) {
+    if (!receiverFileBaseDir.get().exists()) {
+      if (receiverFileBaseDir.get().mkdirs()) {
         LOGGER.info(
             "PipeConsensus-PipeName-{}: Receiver file dir {} was created.",
             consensusPipeName,
-            receiverFileDirWithIdSuffix.get().getPath());
+            receiverFileBaseDir.get().getPath());
       } else {
         LOGGER.error(
             "PipeConsensus-PipeName-{}: Failed to create receiver file dir {}.",
             consensusPipeName,
-            receiverFileDirWithIdSuffix.get().getPath());
+            receiverFileBaseDir.get().getPath());
       }
     }
     // Every tsFileWriter has its own writing path.
@@ -939,23 +940,23 @@ private String getReceiverFileBaseDir() throws DiskSpaceInsufficientException {
 
   private void initiateTsFileBufferFolder() throws DiskSpaceInsufficientException, IOException {
     // Clear the original receiver file dir if exists
-    if (receiverFileDirWithIdSuffix.get() != null) {
-      if (receiverFileDirWithIdSuffix.get().exists()) {
+    if (receiverFileBaseDir.get() != null) {
+      if (receiverFileBaseDir.get().exists()) {
         try {
           RetryUtils.retryOnException(
               () -> {
-                FileUtils.deleteDirectory(receiverFileDirWithIdSuffix.get());
+                FileUtils.deleteDirectory(receiverFileBaseDir.get());
                 return null;
               });
           LOGGER.info(
               "PipeConsensus-PipeName-{}: Original receiver file dir {} was deleted successfully.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath());
+              receiverFileBaseDir.get().getPath());
         } catch (IOException e) {
           LOGGER.warn(
               "PipeConsensus-PipeName-{}: Failed to delete original receiver file dir {}, because {}.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath(),
+              receiverFileBaseDir.get().getPath(),
               e.getMessage(),
               e);
         }
@@ -964,10 +965,10 @@ private void initiateTsFileBufferFolder() throws DiskSpaceInsufficientException,
           LOGGER.debug(
               "PipeConsensus-PipeName-{}: Original receiver file dir {} is not existed. No need to delete.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath());
+              receiverFileBaseDir.get().getPath());
         }
       }
-      receiverFileDirWithIdSuffix.set(null);
+      receiverFileBaseDir.set(null);
     } else {
       LOGGER.debug(
           "PipeConsensus-PipeName-{}: Current receiver file dir is null. No need to delete.",
@@ -1028,7 +1029,7 @@ private void initiateTsFileBufferFolder() throws DiskSpaceInsufficientException,
               "PipeConsensus-PipeName-%s: Failed to create receiver file dir %s. May because authority or dir already exists etc.",
               consensusPipeName, newReceiverDir.getPath()));
     }
-    receiverFileDirWithIdSuffix.set(newReceiverDir);
+    this.receiverFileBaseDir.set(newReceiverDir);
   }
 
   public PipeConsensusRequestVersion getVersion() {
@@ -1040,34 +1041,34 @@ public synchronized void handleExit() {
     pipeConsensusTsFileWriterPool.handleExit(consensusPipeName);
 
     // Clear the original receiver file dir if exists
-    if (receiverFileDirWithIdSuffix.get() != null) {
-      if (receiverFileDirWithIdSuffix.get().exists()) {
+    if (receiverFileBaseDir.get() != null) {
+      if (receiverFileBaseDir.get().exists()) {
         try {
           RetryUtils.retryOnException(
               () -> {
-                FileUtils.deleteDirectory(receiverFileDirWithIdSuffix.get());
+                FileUtils.deleteDirectory(receiverFileBaseDir.get());
                 return null;
               });
           LOGGER.info(
               "PipeConsensus-PipeName-{}: Receiver exit: Original receiver file dir {} was deleted.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath());
+              receiverFileBaseDir.get().getPath());
         } catch (IOException e) {
           LOGGER.warn(
               "PipeConsensus-PipeName-{}: Receiver exit: Delete original receiver file dir {} error.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath(),
+              receiverFileBaseDir.get().getPath(),
               e);
         }
       } else {
         if (LOGGER.isDebugEnabled()) {
           LOGGER.debug(
               "PipeConsensus-PipeName-{}: Receiver exit: Original receiver file dir {} does not exist. No need to delete.",
               consensusPipeName,
-              receiverFileDirWithIdSuffix.get().getPath());
+              receiverFileBaseDir.get().getPath());
         }
       }
-      receiverFileDirWithIdSuffix.set(null);
+      receiverFileBaseDir.set(null);
     } else {
       if (LOGGER.isDebugEnabled()) {
         LOGGER.debug(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java

@@ -407,10 +407,10 @@ private boolean isFileExistedAndNameCorrect(final String fileName) {
     return writingFile != null && writingFile.exists() && writingFile.getName().equals(fileName);
   }
 
-  private void closeCurrentWritingFileWriter(final boolean fsyncAfterClose) {
+  private void closeCurrentWritingFileWriter(final boolean fsyncBeforeClose) {
     if (writingFileWriter != null) {
       try {
-        if (IS_FSYNC_ENABLED && fsyncAfterClose) {
+        if (IS_FSYNC_ENABLED && fsyncBeforeClose) {
           writingFileWriter.getFD().sync();
         }
         writingFileWriter.close();