Repository: tapdata/tapdata (language: python)

Issue:
fix upload PDK slow bug.



Discussion / hints:

Fix patch:
diff --git a/build/image/Dockerfile b/build/image/Dockerfile
index c2a60bdc8b..25a403606c 100644
--- a/build/image/Dockerfile
+++ b/build/image/Dockerfile
@@ -8,9 +8,9 @@ COPY components/ /tapdata/apps/components
 COPY docker-entrypoint.sh /tapdata/apps/start.sh
 COPY bin/ /tapdata/apps
 
-RUN apt install supervisor -y
+RUN apt update && apt install supervisor --fix-missing -y
 
-RUN apt install wget -y && \
+RUN apt install wget --fix-missing -y && \
     if [[ `uname -p` == "aarch64" ]]; then \
       wget https://github.com/async-profiler/async-profiler/releases/download/v3.0/async-profiler-3.0-linux-arm64.tar.gz -O /tapdata/apps/components/async-profiler.tar.gz; \
     else \
diff --git a/manager/tm/src/main/java/com/tapdata/tm/ds/controller/PdkController.java b/manager/tm/src/main/java/com/tapdata/tm/ds/controller/PdkController.java
index 2e17cd6ebe..757ec25047 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/ds/controller/PdkController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/ds/controller/PdkController.java
@@ -9,6 +9,7 @@
 import com.tapdata.tm.ds.vo.PdkFileTypeEnum;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
@@ -30,12 +31,14 @@
 @RestController
 @RequestMapping("/api/pdk")
 @Setter(onMethod_ = {@Autowired})
+@Slf4j
 public class PdkController extends BaseController {
     private PkdSourceService pkdSourceService;
 
     @PostMapping(path = "upload/source", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
     public ResponseMessage<Void> uploadJar(@RequestParam("file") CommonsMultipartFile[] file, @RequestParam("source") List<String> sourceJsons, @RequestParam("latest") boolean latest) {
 
+        log.debug("Process upload pdk source, file size: {}, source size: {}, latest: {}", file.length, sourceJsons.size(), latest);
         List<PdkSourceDto> pdkSourceDtos = new ArrayList<>();
         for (String sourceJson : sourceJsons) {
             // if the size of sourceJsons send by client is 1, here we marshal the data in a weird way, not sure why; so
diff --git a/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/PkdSourceService.java b/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/PkdSourceService.java
index 2f144c44a8..b42b008a34 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/PkdSourceService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/PkdSourceService.java
@@ -2,13 +2,17 @@
 
 import cn.hutool.core.date.DateUtil;
 import com.google.common.collect.Maps;
+import com.mongodb.client.gridfs.GridFSFindIterable;
 import com.mongodb.client.gridfs.model.GridFSFile;
+import com.mongodb.client.result.UpdateResult;
 import com.tapdata.tm.Settings.service.SettingsService;
 import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.commons.schema.DataSourceDefinitionDto;
+import com.tapdata.tm.commons.util.JsonUtil;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.ds.dto.PdkSourceDto;
 import com.tapdata.tm.ds.dto.PdkVersionCheckDto;
+import com.tapdata.tm.ds.repository.PdkSourceRepository;
 import com.tapdata.tm.ds.vo.PdkFileTypeEnum;
 import com.tapdata.tm.file.service.FileService;
 import com.tapdata.tm.tcm.service.TcmService;
@@ -17,6 +21,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -47,6 +52,7 @@ public class PkdSourceService {
 	private FileService fileService;
 	private TcmService tcmService;
 	private SettingsService settingsService;
+	private PdkSourceRepository repository;
 
 	@SuppressWarnings(value = "unchecked")
 	public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSourceDtos, boolean latest, UserDetail user) {
@@ -54,6 +60,9 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 		Map<String, CommonsMultipartFile> docMap = new HashMap<>();
 		CommonsMultipartFile jarFile = null;
 		for (CommonsMultipartFile multipartFile : files) {
+			if (log.isDebugEnabled()) {
+				log.debug("multipartFile name: {}, file size: {}", multipartFile.getOriginalFilename(), multipartFile.getSize());
+			}
 			if (multipartFile.getOriginalFilename() != null && multipartFile.getOriginalFilename().endsWith(".jar")) {
 				jarFile = multipartFile;
 			} else if (multipartFile.getOriginalFilename() != null && multipartFile.getOriginalFilename().endsWith(".md")) {
@@ -95,7 +104,7 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 
 			DataSourceDefinitionDto definitionDto = new DataSourceDefinitionDto();
 			BeanUtils.copyProperties(pdkSourceDto, definitionDto);
-			definitionDto.setId(Objects.nonNull(oldDefinitionDto) ? oldDefinitionDto.getId() : null);
+			//definitionDto.setId(Objects.nonNull(oldDefinitionDto) ? oldDefinitionDto.getId() : null);
 			definitionDto.setConnectionType(pdkSourceDto.getType());
 			definitionDto.setType(pdkSourceDto.getName());
 			definitionDto.setPdkType("pdk");
@@ -109,11 +118,23 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 
 			// remove snapshot overwritten file(jar/icons)
 			if (oldDefinitionDto != null) {
-				fileService.deleteFileById(MongoUtils.toObjectId(oldDefinitionDto.getJarRid()));
+				// change to async delete
+                dataSourceDefinitionService.deleteById(oldDefinitionDto.getId());
+				List<ObjectId> fileIds = new ArrayList<>();
+				fileIds.add(MongoUtils.toObjectId(oldDefinitionDto.getJarRid()));
 				if (oldDefinitionDto.getIcon() != null) {
-					fileService.deleteFileById(MongoUtils.toObjectId(oldDefinitionDto.getIcon()));
+					fileIds.add(MongoUtils.toObjectId(oldDefinitionDto.getIcon()));
 				}
-				fileService.deleteFileByPdkHash(pdkHash, pdkAPIBuildNumber);
+				Query query = Query.query(Criteria.where("metadata.pdkHash").is(pdkHash)
+						.and("metadata.pdkAPIBuildNumber").is(pdkAPIBuildNumber));
+				GridFSFindIterable result = fileService.find(query);
+				result.forEach(gridFSFile ->
+					fileIds.add(gridFSFile.getObjectId())
+				);
+
+				fileService.scheduledDeleteFiles(fileIds, "Upload new connector", "DatabaseTypes",
+						oldDefinitionDto.getId(), user);
+				log.debug("Delete original source {}", oldDefinitionDto.getId());
 			}
 
 			// upload the associated files(jar/icons)
@@ -128,6 +149,7 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 
 				// 1. upload jar file, only update once
 				jarObjectId = fileService.storeFile(jarFile.getInputStream(), jarFile.getOriginalFilename(), null, fileInfo);
+
 				// 2. upload the associated icon
 				CommonsMultipartFile icon = iconMap.getOrDefault(pdkSourceDto.getIcon(), null);
 				if (icon != null) {
@@ -135,6 +157,7 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 				}
 				// 3. upload readeMe doc
                 uploadDocs(docMap, pdkSourceDto.getMessages(), fileInfo, oemConfig);
+                log.debug("Upload file to GridFS success");
 			} catch (IOException e) {
 				throw new BizException("SystemError", e);
 			}
@@ -163,6 +186,7 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 			} else {
 				dataSourceDefinitionService.upsert(Query.query(Criteria.where("_id").is(definitionDto.getId())), definitionDto, user);
 			}
+			log.debug("Upsert data source definition success");
 
 			//根据数据源类型删除可能存在的旧的pdk
 			FunctionUtils.ignoreAnyError(() -> {
@@ -179,6 +203,7 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 			});
 
 		}
+		log.debug("Upload pdk done.");
 	}
 	public String checkJarMD5(String pdkHash, int pdkBuildNumber){
 		String md5 = null;
diff --git a/manager/tm/src/main/java/com/tapdata/tm/file/entity/WaitingDeleteFile.java b/manager/tm/src/main/java/com/tapdata/tm/file/entity/WaitingDeleteFile.java
new file mode 100644
index 0000000000..4577711a24
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/file/entity/WaitingDeleteFile.java
@@ -0,0 +1,25 @@
+package com.tapdata.tm.file.entity;
+
+import com.tapdata.tm.base.entity.BaseEntity;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.util.List;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/8/7 11:59
+ */
+@EqualsAndHashCode(callSuper = true)
+@Document("WaitingDeleteFile")
+@Data
+public class WaitingDeleteFile extends BaseEntity {
+
+    private List<ObjectId> fileIds;
+    private String reason;
+    private String module;
+    private ObjectId resourceId;
+
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepository.java b/manager/tm/src/main/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepository.java
new file mode 100644
index 0000000000..779484950d
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepository.java
@@ -0,0 +1,19 @@
+package com.tapdata.tm.file.repository;
+
+import com.tapdata.tm.base.reporitory.BaseRepository;
+import com.tapdata.tm.file.entity.WaitingDeleteFile;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.stereotype.Repository;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/8/7 12:02
+ */
+@Repository
+public class WaitingDeleteFileRepository extends BaseRepository<WaitingDeleteFile, ObjectId> {
+
+    public WaitingDeleteFileRepository(MongoTemplate mongoOperations) {
+        super(WaitingDeleteFile.class, mongoOperations);
+    }
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/file/service/FileService.java b/manager/tm/src/main/java/com/tapdata/tm/file/service/FileService.java
index 991000c275..ca2bd437a3 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/file/service/FileService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/file/service/FileService.java
@@ -8,14 +8,20 @@
 import com.mongodb.client.gridfs.GridFSFindIterable;
 import com.mongodb.client.gridfs.model.GridFSFile;
 import com.tapdata.tm.base.exception.BizException;
+import com.tapdata.tm.config.security.UserDetail;
+import com.tapdata.tm.file.entity.WaitingDeleteFile;
+import com.tapdata.tm.file.repository.WaitingDeleteFileRepository;
 import com.tapdata.tm.utils.FunctionUtils;
 import com.tapdata.tm.utils.GZIPUtil;
 import com.tapdata.tm.commons.util.ThrowableUtils;
+import io.jsonwebtoken.lang.Collections;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.compress.utils.IOUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.bson.Document;
 import org.bson.types.ObjectId;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.mongodb.MongoDatabaseFactory;
@@ -31,6 +37,7 @@
 import java.io.*;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -43,6 +50,9 @@
 @Slf4j
 public class FileService {
 
+    @Autowired
+    @Setter
+    private WaitingDeleteFileRepository waitingDeleteFileRepository;
     @Resource
     private MongoDatabaseFactory mongoDatabaseFactory;
     @Bean
@@ -58,6 +68,7 @@ public GridFSBucket getGridFSBuckets() {
             };
 
     @Resource
+    @Setter
     private GridFSBucket gridFSBucket;
     private final GridFsTemplate gridFsTemplate;
     public FileService(GridFsTemplate gridFsTemplate){
@@ -240,4 +251,34 @@ public void viewWord(XWPFTemplate template, HttpServletResponse response, String
             log.error("viewWord error {}", ThrowableUtils.getStackTraceByPn(e));
         }
     }
+
+    /**
+     * schedule delete files by id
+     * @param fileIds file ids
+     * @param reason reason
+     * @param module module
+     * @param resourceId original resource id
+     * @param userDetail user detail
+     */
+    public void scheduledDeleteFiles(List<ObjectId> fileIds, String reason, String module, ObjectId resourceId, UserDetail userDetail) {
+        WaitingDeleteFile entity = new WaitingDeleteFile();
+        entity.setFileIds(fileIds);
+        entity.setReason(reason);
+        entity.setModule(module);
+        entity.setResourceId(resourceId);
+        waitingDeleteFileRepository.insert(entity, userDetail);
+    }
+
+    /**
+     * execute cleanup
+     */
+    public void cleanupWaitingDeleteFiles() {
+        List<WaitingDeleteFile> result = waitingDeleteFileRepository.findAll(Query.query(new Criteria()));
+        result.forEach(waitingDeleteFile -> {
+            if (!Collections.isEmpty(waitingDeleteFile.getFileIds())) {
+                waitingDeleteFile.getFileIds().forEach(this::deleteFileById);
+            }
+            waitingDeleteFileRepository.deleteById(waitingDeleteFile.getId());
+        });
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/schedule/DeleteFileSchedule.java b/manager/tm/src/main/java/com/tapdata/tm/schedule/DeleteFileSchedule.java
new file mode 100644
index 0000000000..fb281645a3
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/schedule/DeleteFileSchedule.java
@@ -0,0 +1,31 @@
+package com.tapdata.tm.schedule;
+
+import com.tapdata.tm.file.service.FileService;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/8/2 16:48
+ */
+@Slf4j
+@Component
+public class DeleteFileSchedule {
+    @Autowired
+    @Setter
+    private FileService fileService;
+
+    /**
+     * clean up files that need to be deleted
+     */
+    @Scheduled(cron = "0 0 2 * * ?")
+    @SchedulerLock(name = "DatabaseTypeSchedule.cleanUpForDatabaseTypes", lockAtMostFor = "PT1H", lockAtLeastFor = "PT1H")
+    public void cleanUpForDatabaseTypes() {
+        log.info("clean up files that need to be deleted");
+        fileService.cleanupWaitingDeleteFiles();
+    }
+}
diff --git a/tapdata-cli/src/main/java/io/tapdata/pdk/cli/services/UploadFileService.java b/tapdata-cli/src/main/java/io/tapdata/pdk/cli/services/UploadFileService.java
index 3b58882e7c..424f42e31e 100644
--- a/tapdata-cli/src/main/java/io/tapdata/pdk/cli/services/UploadFileService.java
+++ b/tapdata-cli/src/main/java/io/tapdata/pdk/cli/services/UploadFileService.java
@@ -181,6 +181,7 @@ public static void upload(Map<String, InputStream> inputStreamMap, File file, Li
       request = new HttpRequest(url, method);
     }
     request.connectTimeout(180000).readTimeout(180000);//连接超时设置
+    //request.progress((uploaded, total) -> System.out.println("uploaded: " + uploaded + " total: " + total + " time: " + System.currentTimeMillis()));
     if (file != null) {
       request.part("file", file.getName(), "application/java-archive", file);
     }


Test patch:
diff --git a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/PkdSourceServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/PkdSourceServiceTest.java
index cec7bc67a2..4bed7c2fff 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/PkdSourceServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/PkdSourceServiceTest.java
@@ -1,20 +1,27 @@
 package com.tapdata.tm.ds.service.impl;
 
+import cn.hutool.core.date.DateUtil;
+import com.mongodb.client.gridfs.GridFSFindIterable;
 import com.mongodb.client.gridfs.model.GridFSFile;
 import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.commons.schema.DataSourceDefinitionDto;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.ds.dto.PdkSourceDto;
+import com.tapdata.tm.ds.dto.PdkVersionCheckDto;
+import com.tapdata.tm.ds.vo.PdkFileTypeEnum;
 import com.tapdata.tm.file.service.FileService;
+import com.tapdata.tm.tcm.service.TcmService;
 import com.tapdata.tm.utils.MessageUtil;
 import lombok.SneakyThrows;
 import org.apache.commons.fileupload.FileItem;
 import org.apache.commons.io.FileUtils;
 import org.bson.Document;
 import org.bson.types.ObjectId;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -25,6 +32,7 @@
 import java.io.File;
 import java.io.InputStream;
 import java.util.*;
+import java.util.function.Consumer;
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
@@ -67,6 +75,30 @@ void testUploadPdk(){
             pkdSourceService.uploadPdk(files,pdkSourceDtos,latest,user);
             verify(fileService).storeFile(any(),anyString(),any(),anyMap());
             FileUtils.deleteQuietly(new File("a.jar"));
+
+            GridFSFindIterable result = mock(GridFSFindIterable.class);
+
+            DataSourceDefinitionDto dto = new DataSourceDefinitionDto();
+            dto.setId(new ObjectId());
+            dto.setJarRid(new ObjectId().toHexString());
+            dto.setIcon(new ObjectId().toHexString());
+            when(dataSourceDefinitionService.findOne(any())).thenReturn(dto);
+            when(fileService.find(any())).thenReturn(result);
+            doAnswer(answer -> {
+
+                Consumer consumer = answer.getArgument(0);
+                GridFSFile gridFSFile = mock(GridFSFile.class);
+                when(gridFSFile.getObjectId()).thenReturn(new ObjectId());
+                consumer.accept(gridFSFile);
+
+                return null;
+            }).when(result).forEach(any());
+            pkdSourceService.uploadPdk(files, pdkSourceDtos, true, user);
+            verify(fileService).scheduledDeleteFiles(any(), anyString(), anyString(), any(), any());
+
+            /*pkdSourceService.uploadPdk(files, pdkSourceDtos, true, user);
+            verify(fileService).scheduledDeleteFiles(any(), anyString(), anyString(), any(), any());*/
+
         }
     }
     @Nested
@@ -280,4 +312,98 @@ void testSuccess() {
             });
         }
     }
+
+    @Test
+    public void testVersionCheck() {
+        TcmService tcmService = mock(TcmService.class);
+        List<DataSourceDefinitionDto> result = new ArrayList<>();
+        when(dataSourceDefinitionService.findAll(any(Query.class))).thenReturn(result);
+        when(tcmService.getLatestProductReleaseCreateTime()).thenReturn("2023-04-28");
+
+        pkdSourceService.setTcmService(tcmService);
+
+        Assertions.assertDoesNotThrow(() -> {
+            List<PdkVersionCheckDto> a = pkdSourceService.versionCheck(5);
+            Assertions.assertNotNull(a);
+            Assertions.assertTrue(a.isEmpty());
+        });
+
+        for (int i = 0; i < 5; i++) {
+            DataSourceDefinitionDto dto = new DataSourceDefinitionDto();
+            dto.setPdkId("test");
+            dto.setPdkAPIBuildNumber(i);
+            dto.setPdkAPIVersion("api_v" + i);
+            dto.setPdkHash("123" + i);
+            dto.setLastUpdAt(new Date());
+            result.add(dto);
+        }
+
+        List<PdkVersionCheckDto> versionCheckResult = pkdSourceService.versionCheck(2);
+        Assertions.assertEquals(1, versionCheckResult.size());
+        Assertions.assertEquals("1234", versionCheckResult.get(0).getPdkHash());
+        Assertions.assertEquals(DateUtil.formatDateTime(result.get(4).getLastUpdAt()), versionCheckResult.get(0).getGitBuildTime());
+        Assertions.assertTrue(versionCheckResult.get(0).isLatest());
+
+        Map<String, String> manifest = new HashMap<>();
+        manifest.put("Git-Build-Time", "2023-04-25T18:05:20+0800");
+        result.get(4).setManifest(manifest);
+
+        versionCheckResult = pkdSourceService.versionCheck(2);
+
+        Assertions.assertEquals(1, versionCheckResult.size());
+        Assertions.assertEquals("1234", versionCheckResult.get(0).getPdkHash());
+        Assertions.assertEquals("2023-04-25 18:05:20", versionCheckResult.get(0).getGitBuildTime());
+        Assertions.assertTrue(versionCheckResult.get(0).isLatest());
+    }
+
+    @Test
+    public void testUploadAndView() {
+
+        String pdkHash = "test";
+        int pdkBuildNumber = 1;
+        UserDetail user = mock(UserDetail.class);
+        PdkFileTypeEnum type = PdkFileTypeEnum.JAR;
+        HttpServletResponse response = mock(HttpServletResponse.class);
+
+        Assertions.assertDoesNotThrow(() -> {
+
+            when(dataSourceDefinitionService.findOne(any(Query.class))).thenReturn(null);
+            ArgumentCaptor<Integer> captor = ArgumentCaptor.forClass(Integer.class);
+            doNothing().when(response).sendError(captor.capture());
+            pkdSourceService.uploadAndView(pdkHash, pdkBuildNumber, user, type, response);
+
+            Assertions.assertEquals(404, captor.getValue());
+            DataSourceDefinitionDto dto = new DataSourceDefinitionDto();
+            dto.setScope("customer");
+            dto.setCustomId("customer_id");
+            when(user.getCustomerId()).thenReturn("customer_id_1");
+            when(dataSourceDefinitionService.findOne(any(Query.class))).thenReturn(dto);
+            Assertions.assertThrows(BizException.class, () -> {
+                pkdSourceService.uploadAndView(pdkHash, pdkBuildNumber, user, PdkFileTypeEnum.IMAGE, response);
+            });
+
+            when(user.getCustomerId()).thenReturn("customer_id");
+            when(dataSourceDefinitionService.findOne(any(Query.class))).thenReturn(dto);
+            pkdSourceService.uploadAndView(pdkHash, pdkBuildNumber, user, PdkFileTypeEnum.MARKDOWN, response);
+            Assertions.assertTrue(captor.getAllValues().stream().anyMatch(p -> p == 404));
+
+            dto.setMessages(new LinkedHashMap<>());
+            dto.getMessages().put("zh_CN", new HashMap<String, String>(){{
+                put("doc", new ObjectId().toHexString());
+            }});
+            dto.getMessages().put("en_US", new HashMap<String, String>(){{
+                put("doc", new ObjectId().toHexString());
+            }});
+            dto.getMessages().put("zh_TW", new HashMap<String, String>(){{
+                put("doc", new ObjectId().toHexString());
+            }});
+
+            pkdSourceService.uploadAndView(pdkHash, pdkBuildNumber, user, PdkFileTypeEnum.MARKDOWN, response);
+            verify(fileService).viewImg(any(), any());
+
+        });
+
+
+
+    }
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepositoryTest.java b/manager/tm/src/test/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepositoryTest.java
new file mode 100644
index 0000000000..55cc251bcb
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/file/repository/WaitingDeleteFileRepositoryTest.java
@@ -0,0 +1,32 @@
+package com.tapdata.tm.file.repository;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * 
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/9/30 14:03
+ */public class WaitingDeleteFileRepositoryTest {
+
+    @Test
+    public void testWaitingDeleteFileRepository() {
+        MongoTemplate mongoTemplate = mock(MongoTemplate.class);
+        MongoConverter converter = mock(MongoConverter.class);
+        when(mongoTemplate.getConverter()).thenReturn(converter);
+        MappingContext mappingContext = mock(MappingContext.class);
+        when(converter.getMappingContext()).thenReturn(mappingContext);
+        MongoPersistentEntity entity = mock(MongoPersistentEntity.class);
+        when(mappingContext.getRequiredPersistentEntity(any(Class.class))).thenReturn(entity);
+        WaitingDeleteFileRepository repository = new WaitingDeleteFileRepository(mongoTemplate);
+        assertNotNull( repository);
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/file/service/FileServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/file/service/FileServiceTest.java
new file mode 100644
index 0000000000..b4ea7dc7ff
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/file/service/FileServiceTest.java
@@ -0,0 +1,384 @@
+package com.tapdata.tm.file.service;
+
+import com.deepoove.poi.XWPFTemplate;
+import com.mongodb.Function;
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.MongoIterable;
+import com.mongodb.client.gridfs.GridFSBucket;
+import com.mongodb.client.gridfs.GridFSFindIterable;
+import com.mongodb.client.gridfs.model.GridFSFile;
+import com.mongodb.client.model.Collation;
+import com.tapdata.tm.config.security.UserDetail;
+import com.tapdata.tm.file.entity.WaitingDeleteFile;
+import com.tapdata.tm.file.repository.WaitingDeleteFileRepository;
+import com.tapdata.tm.utils.GZIPUtil;
+import org.apache.commons.io.IOUtils;
+import org.bson.conversions.Bson;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.gridfs.GridFsResource;
+import org.springframework.data.mongodb.gridfs.GridFsTemplate;
+
+import javax.servlet.ServletOutputStream;
+import javax.servlet.WriteListener;
+import javax.servlet.http.HttpServletResponse;
+import java.io.*;
+import java.nio.charset.Charset;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/9/29 14:44
+ */
+public class FileServiceTest {
+
+    @Test
+    public void testFind() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        GridFSFindIterable result = mock(GridFSFindIterable.class);
+
+        when(gridFsTemplate.find(any())).thenReturn(result);
+        assertDoesNotThrow(() -> {
+            GridFSFindIterable resultSet = fileService.find(Query.query(Criteria.where("filename").is("test")));
+            Assertions.assertNotNull(resultSet);
+        });
+    }
+
+    @Test
+    public void testFindOne() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        GridFSFile gridFSFile = mock(GridFSFile.class);
+
+        when(gridFsTemplate.findOne(any())).thenReturn(gridFSFile);
+        assertDoesNotThrow(() -> {
+            GridFSFile result = fileService.findOne(Query.query(Criteria.where("filename").is("test")));
+            Assertions.assertNotNull(result);
+        });
+    }
+
+    @Test
+    public void testStoreFile() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+
+        ObjectId id = new ObjectId();
+
+        when(gridFsTemplate.store(any(), any(), any(), any())).then(ans -> {
+
+            Assertions.assertEquals("test.png", ans.getArgument(1));
+
+            return id;
+        });
+        assertDoesNotThrow(() -> {
+            InputStream inputStream = new ByteArrayInputStream("".getBytes(Charset.defaultCharset()));
+
+            ObjectId result = fileService.storeFile(inputStream, "test.png", "image/png", new HashMap<>());
+            Assertions.assertNotNull(result);
+            Assertions.assertEquals(id, result);
+        });
+    }
+
+    @Test
+    public void testReadFileById() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+
+        InputStream inputStream = new ByteArrayInputStream("test".getBytes(Charset.defaultCharset()));
+        GridFsResource gridFsResource = mock(GridFsResource.class);
+        GridFSFile gridFSFile = mock(GridFSFile.class);
+
+        Assertions.assertDoesNotThrow(() -> {
+            when(gridFsTemplate.getResource(any(GridFSFile.class))).thenReturn(gridFsResource);
+            when(gridFsResource.getInputStream()).thenReturn(inputStream);
+
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+            long length = fileService.readFileById(gridFSFile, outputStream);
+
+            Assertions.assertEquals(4, length);
+            String content = new String(outputStream.toByteArray(), Charset.defaultCharset());
+            Assertions.assertEquals("test", content);
+
+            outputStream.reset();
+            when(gridFsResource.getInputStream()).thenThrow(new IOException("test"));
+            long result = fileService.readFileById(gridFSFile, outputStream);
+            assertEquals(0, result);
+
+        });
+    }
+
+    @Test
+    public void testReadFileByIdToResponse() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+
+        InputStream inputStream = new ByteArrayInputStream("test".getBytes(Charset.defaultCharset()));
+        GridFsResource gridFsResource = mock(GridFsResource.class);
+        GridFSFile gridFSFile = mock(GridFSFile.class);
+        HttpServletResponse response = mock(HttpServletResponse.class);
+
+        ObjectId id = new ObjectId();
+
+        Assertions.assertDoesNotThrow(() -> {
+
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+
+            when(gridFsTemplate.findOne(any())).thenReturn(gridFSFile);
+            when(gridFsTemplate.getResource(gridFSFile)).thenReturn(gridFsResource);
+            when(gridFsResource.getInputStream()).thenReturn(inputStream);
+            when(gridFsResource.getFilename()).thenReturn("test.png");
+            when(response.getOutputStream()).thenReturn(new ServletOutputStream() {
+                @Override
+                public boolean isReady() {
+                    return true;
+                }
+
+                @Override
+                public void setWriteListener(WriteListener listener) {
+
+                }
+
+                @Override
+                public void write(int b) throws IOException {
+                    outputStream.write(b);
+                }
+            });
+            doAnswer(answer -> {
+                Assertions.assertEquals("application/octet-stream", answer.getArgument(0));
+                return null;
+            }).when(response).setContentType(any());
+
+            long length = fileService.readFileById(id, response);
+
+            Assertions.assertEquals(4, length);
+            Assertions.assertEquals("test", new String(outputStream.toByteArray(), Charset.defaultCharset()));
+
+            outputStream.reset();
+            when(gridFsResource.getInputStream()).thenThrow(new IOException("test"));
+            long result = fileService.readFileById(gridFSFile, outputStream);
+            assertEquals(0, result);
+        });
+    }
+
+    @Test
+    public void testDeleteFileById() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        doNothing().when(gridFsTemplate).delete(any());
+        Assertions.assertDoesNotThrow(() -> {
+            fileService.deleteFileById(new ObjectId());
+        });
+    }
+
+    @Test
+    public void testDeleteFileByPdkHash() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        doNothing().when(gridFsTemplate).delete(any());
+        Assertions.assertDoesNotThrow(() -> {
+            fileService.deleteFileByPdkHash("test", 2);
+        });
+    }
+
+    @Test
+    public void testViewImg() {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        GridFSBucket gridFSBucket = mock(GridFSBucket.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        fileService.setGridFSBucket(gridFSBucket);
+        GridFSFile gridFSFile = mock(GridFSFile.class);
+        HttpServletResponse response = mock(HttpServletResponse.class);
+        ObjectId fileId = new ObjectId();
+
+        Assertions.assertDoesNotThrow(() -> {
+
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+
+            when(gridFsTemplate.findOne(any())).thenReturn(gridFSFile);
+            when(gridFSFile.getFilename()).thenReturn("test.png");
+            when(response.getOutputStream()).thenReturn(new ServletOutputStream() {
+                @Override
+                public boolean isReady() {
+                    return true;
+                }
+
+                @Override
+                public void setWriteListener(WriteListener listener) {
+
+                }
+
+                @Override
+                public void write(int b) throws IOException {
+                    outputStream.write(b);
+                }
+            });
+
+            doAnswer(answer -> {
+                OutputStream output = answer.getArgument(1);
+                output.write("test".getBytes(Charset.defaultCharset()));
+                output.flush();
+                return null;
+            }).when(gridFSBucket).downloadToStream(any(ObjectId.class), any(OutputStream.class));
+
+            fileService.viewImg(fileId, response);
+
+            Assertions.assertEquals("test", new String(outputStream.toByteArray(), Charset.defaultCharset()));
+
+        });
+
+    }
+
+    @Test
+    public void testViewImg1() {
+
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        HttpServletResponse response = mock(HttpServletResponse.class);
+        Assertions.assertDoesNotThrow(() -> {
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+            when(response.getOutputStream()).thenReturn(new ServletOutputStream() {
+                @Override
+                public boolean isReady() {
+                    return true;
+                }
+
+                @Override
+                public void setWriteListener(WriteListener listener) {
+
+                }
+
+                @Override
+                public void write(int b) throws IOException {
+                    outputStream.write(b);
+                }
+            });
+
+            String fileData = "{test: 1}";
+            fileService.viewImg1(fileData, response, "test.json");
+
+            Assertions.assertEquals(fileData, new String(Objects.requireNonNull(GZIPUtil.unGzip(outputStream.toByteArray())), Charset.defaultCharset()));
+        });
+
+    }
+
+    @Test
+    public void testViewWord () {
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        HttpServletResponse response = mock(HttpServletResponse.class);
+        Assertions.assertDoesNotThrow(() -> {
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+            when(response.getOutputStream()).thenReturn(new ServletOutputStream() {
+                @Override
+                public boolean isReady() {
+                    return true;
+                }
+
+                @Override
+                public void setWriteListener(WriteListener listener) {
+
+                }
+
+                @Override
+                public void write(int b) throws IOException {
+                    outputStream.write(b);
+                }
+            });
+
+            XWPFTemplate xwpfTemplate = mock(XWPFTemplate.class);
+            doAnswer(answer -> {
+
+                OutputStream output = answer.getArgument(0);
+                output.write("test".getBytes(Charset.defaultCharset()));
+                output.flush();
+                return null;
+            }).when(xwpfTemplate).writeAndClose(any(OutputStream.class));
+
+            fileService.viewWord(xwpfTemplate, response, "test.xml");
+
+            Assertions.assertEquals("test", new String(Objects.requireNonNull(outputStream.toByteArray()), Charset.defaultCharset()));
+
+            outputStream.reset();
+
+            fileService.viewWord(xwpfTemplate, response, null);
+            Assertions.assertEquals("test", new String(Objects.requireNonNull(outputStream.toByteArray()), Charset.defaultCharset()));
+        });
+    }
+
+    @Test
+    public void testScheduledDeleteFiles() {
+
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+
+        WaitingDeleteFileRepository waitingDeleteFileRepository = mock(WaitingDeleteFileRepository.class);
+
+        FileService fileService = new FileService(gridFsTemplate);
+        fileService.setWaitingDeleteFileRepository(waitingDeleteFileRepository);
+
+        List<ObjectId> fileIds = new ArrayList<>();
+        fileIds.add(new ObjectId());
+        fileIds.add(new ObjectId());
+        fileIds.add(new ObjectId());
+
+        String reason = "Upload new connector";
+        String module = "connector";
+        ObjectId resourceId = new ObjectId();
+        UserDetail userDetail = mock(UserDetail.class);
+
+        when(waitingDeleteFileRepository.insert(any(WaitingDeleteFile.class), any())).then(answer -> {
+            Object obj = answer.getArgument(0);
+            Assertions.assertTrue(obj instanceof WaitingDeleteFile);
+            WaitingDeleteFile entity = (WaitingDeleteFile) obj;
+            Assertions.assertEquals(fileIds.size(), entity.getFileIds().size());
+            Assertions.assertEquals(reason, entity.getReason());
+            Assertions.assertEquals(module, entity.getModule());
+            Assertions.assertEquals(resourceId, entity.getResourceId());
+            return entity;
+        });
+
+        assertDoesNotThrow(() -> {
+
+            fileService.scheduledDeleteFiles(fileIds, reason, module, resourceId, userDetail);
+
+        });
+
+    }
+
+    @Test
+    public void testCleanupWaitingDeleteFiles() {
+        WaitingDeleteFileRepository waitingDeleteFileRepository = mock(WaitingDeleteFileRepository.class);
+        GridFsTemplate gridFsTemplate = mock(GridFsTemplate.class);
+        FileService fileService = new FileService(gridFsTemplate);
+        fileService.setWaitingDeleteFileRepository(waitingDeleteFileRepository);
+
+        ArgumentCaptor<Query> captor = ArgumentCaptor.forClass(Query.class);
+        doNothing().when(gridFsTemplate).delete(captor.capture());
+
+        List<WaitingDeleteFile> result = new ArrayList<>();
+        result.add(new WaitingDeleteFile());
+        result.get(0).setFileIds(Arrays.asList(new ObjectId(), new ObjectId()));
+        when(waitingDeleteFileRepository.findAll(any(Query.class))).thenReturn(result);
+
+        fileService.cleanupWaitingDeleteFiles();
+
+        Assertions.assertEquals(2, captor.getAllValues().size());
+        Assertions.assertEquals(result.get(0).getFileIds().get(0), captor.getAllValues().get(0).getQueryObject().get("_id"));
+        Assertions.assertEquals(result.get(0).getFileIds().get(1), captor.getAllValues().get(1).getQueryObject().get("_id"));
+
+    }
+
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/schedule/DeleteFileScheduleTest.java b/manager/tm/src/test/java/com/tapdata/tm/schedule/DeleteFileScheduleTest.java
new file mode 100644
index 0000000000..22a7fb67b4
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/schedule/DeleteFileScheduleTest.java
@@ -0,0 +1,109 @@
+package com.tapdata.tm.schedule;
+
+import com.mongodb.ServerAddress;
+import com.mongodb.ServerCursor;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
+import com.tapdata.tm.commons.schema.DataSourceDefinitionDto;
+import com.tapdata.tm.ds.repository.DataSourceDefinitionRepository;
+import com.tapdata.tm.ds.service.impl.PkdSourceService;
+import com.tapdata.tm.file.entity.WaitingDeleteFile;
+import com.tapdata.tm.file.repository.WaitingDeleteFileRepository;
+import com.tapdata.tm.file.service.FileService;
+import org.bson.Document;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/8/2 18:42
+ */
+@ExtendWith(MockitoExtension.class)
+public class DeleteFileScheduleTest {
+
+    @Mock
+    private FileService fileService;
+    @Mock
+    private WaitingDeleteFileRepository waitingDeleteFileRepository;
+
+    @Test
+    public void testCleanUpForDatabaseTypes() {
+
+        assertNotNull(fileService);
+        assertNotNull(waitingDeleteFileRepository);
+        fileService.setWaitingDeleteFileRepository(waitingDeleteFileRepository);
+
+        DeleteFileSchedule deleteFileSchedule = new DeleteFileSchedule();
+        deleteFileSchedule.setFileService(fileService);
+
+        assertDoesNotThrow(deleteFileSchedule::cleanUpForDatabaseTypes);
+
+        verify(fileService, times(1)).cleanupWaitingDeleteFiles();
+
+    }
+
+    private static class MockMongoCursor<T> implements MongoCursor<T> {
+        private final List<T> result;
+
+        public MockMongoCursor(List<T> result) {
+            this.result = result;
+        }
+
+        private int readIndex = -1;
+        @Override
+        public void close() {
+
+        }
+
+        @Override
+        public boolean hasNext() {
+            return result.size() > readIndex + 1;
+        }
+
+        @Override
+        public T next() {
+            readIndex++;
+            return result.get(readIndex);
+        }
+
+        @Override
+        public int available() {
+            return 0;
+        }
+
+        @Override
+        public T tryNext() {
+            if (hasNext()) {
+                return next();
+            }
+            return null;
+        }
+
+        @Override
+        public ServerCursor getServerCursor() {
+            return null;
+        }
+
+        @Override
+        public ServerAddress getServerAddress() {
+            return null;
+        }
+
+    }
+}