Fix memory leak of S3 multipart writer in data_backup logics

src/storage/backup/ob_backup_ctx.cpp

@@ -280,7 +280,15 @@ ObBackupDataCtx::ObBackupDataCtx()
 {}
 
 ObBackupDataCtx::~ObBackupDataCtx()
-{}
+{
+  int ret = OB_SUCCESS;
+  if (OB_NOT_NULL(dev_handle_) && io_fd_.is_valid()) {
+    ObBackupIoAdapter util;
+    if (OB_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+      LOG_WARN("fail to close device and fd", K(ret), K_(dev_handle), K_(io_fd));
+    }
+  }
+}
 
 int ObBackupDataCtx::open(const ObLSBackupDataParam &param, const share::ObBackupDataType &backup_data_type,
     const int64_t file_id)
@@ -383,6 +391,7 @@ int ObBackupDataCtx::close()
 {
   int ret = OB_SUCCESS;
   int tmp_ret = OB_SUCCESS;
+  ObBackupIoAdapter util;
   if (IS_NOT_INIT) {
     ret = OB_NOT_INIT;
     LOG_WARN("backup data ctx do not init", K(ret));
@@ -393,6 +402,13 @@ int ObBackupDataCtx::close()
   } else if (OB_FAIL(file_write_ctx_.close())) {
     LOG_WARN("failed to close file writer", K(ret));
   }
+  if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+    ret = COVER_SUCC(tmp_ret);
+    LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
+  } else {
+    dev_handle_ = NULL;
+    io_fd_.reset();
+  }
   return ret;
 }
 

src/storage/backup/ob_backup_extern_info_mgr.cpp

@@ -433,6 +433,8 @@ int ObExternTabletMetaWriter::switch_file_()
 int ObExternTabletMetaWriter::close()
 {
   int ret = OB_SUCCESS;
+  int tmp_ret = OB_SUCCESS;
+  ObBackupIoAdapter util;
   if (IS_NOT_INIT) {
     ret = OB_NOT_INIT;
     LOG_WARN("tablet meta writer not inited", K(ret));
@@ -441,6 +443,13 @@ int ObExternTabletMetaWriter::close()
   } else if (OB_FAIL(file_write_ctx_.close())) {
     LOG_WARN("failed to close file writer", K(ret));
   }
+  if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+    ret = COVER_SUCC(tmp_ret);
+    LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
+  } else {
+    dev_handle_ = NULL;
+    io_fd_.reset();
+  }
   return ret;
 }
 

src/storage/backup/ob_backup_file_writer_ctx.cpp

@@ -132,8 +132,6 @@ int ObBackupFileWriteCtx::commit_file_()
     LOG_WARN("dev handle should not be null", K(ret));
   } else if (OB_FAIL(flush_buffer_(true /*is_last_part*/))) {
     LOG_WARN("failed to flush buffer", K(ret));
-  } else if (OB_FAIL(dev_handle_->close(io_fd_))) {
-    LOG_WARN("failed to close file", K(ret), K_(io_fd));
   } else {
     LOG_INFO("backup file write ctx commit file");
   }

src/storage/backup/ob_backup_index_merger.cpp

@@ -551,7 +551,15 @@ ObIBackupIndexMerger::ObIBackupIndexMerger()
 {}
 
 ObIBackupIndexMerger::~ObIBackupIndexMerger()
-{}
+{
+  int ret = OB_SUCCESS;
+  if (OB_NOT_NULL(dev_handle_) && io_fd_.is_valid()) {
+    ObBackupIoAdapter util;
+    if (OB_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+      LOG_WARN("fail to close device and fd", K(ret), K_(dev_handle), K_(io_fd));
+    }
+  }
+}
 
 int ObIBackupIndexMerger::get_all_retries_(const int64_t task_id, const uint64_t tenant_id,
     const share::ObBackupDataType &backup_data_type, const share::ObLSID &ls_id, common::ObISQLClient &sql_client,
@@ -753,6 +761,8 @@ void ObBackupMacroBlockIndexMerger::reset()
 int ObBackupMacroBlockIndexMerger::merge_index()
 {
   int ret = OB_SUCCESS;
+  int tmp_ret = OB_SUCCESS;
+  ObBackupIoAdapter util;
   ObIBackupMacroBlockIndexFuser *fuser = NULL;
   MERGE_ITER_ARRAY unfinished_iters;
   MERGE_ITER_ARRAY min_iters;
@@ -808,6 +818,13 @@ int ObBackupMacroBlockIndexMerger::merge_index()
   if (OB_NOT_NULL(fuser)) {
     ObLSBackupFactory::free(fuser);
   }
+  if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+    ret = COVER_SUCC(tmp_ret);
+    LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
+  } else {
+    dev_handle_ = NULL;
+    io_fd_.reset();
+  }
   return ret;
 }
 
@@ -1264,6 +1281,8 @@ void ObBackupMetaIndexMerger::reset()
 int ObBackupMetaIndexMerger::merge_index()
 {
   int ret = OB_SUCCESS;
+  int tmp_ret = OB_SUCCESS;
+  ObBackupIoAdapter util;
   MERGE_ITER_ARRAY unfinished_iters;
   MERGE_ITER_ARRAY min_iters;
   if (IS_NOT_INIT) {
@@ -1310,6 +1329,13 @@ int ObBackupMetaIndexMerger::merge_index()
       }
     }
   }
+  if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
+    ret = COVER_SUCC(tmp_ret);
+    LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
+  } else {
+    dev_handle_ = NULL;
+    io_fd_.reset();
+  }
   return ret;
 }