Repository: tapdata/tapdata (language: python)

Issue:
fix E11000 duplicate key error on upload jar



Discussion / hints:

Fix patch:
diff --git a/RELEASE b/RELEASE
index 4715efa8ec..f40a0d9ac7 100644
--- a/RELEASE
+++ b/RELEASE
@@ -1,2 +1,2 @@
-LST: v3.13.0
-DEV: v3.14.0-dev
\ No newline at end of file
+LST: v3.14.0
+DEV: v3.15.0-dev
\ No newline at end of file
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/iengine/api/src/main/java/io/tapdata/flow/engine/V2/sharecdc/ShareCdcContext.java b/iengine/api/src/main/java/io/tapdata/flow/engine/V2/sharecdc/ShareCdcContext.java
index 29a916f3e6..dfe13a1206 100644
--- a/iengine/api/src/main/java/io/tapdata/flow/engine/V2/sharecdc/ShareCdcContext.java
+++ b/iengine/api/src/main/java/io/tapdata/flow/engine/V2/sharecdc/ShareCdcContext.java
@@ -4,7 +4,6 @@
 import io.tapdata.loglistener.TapLogger;
 
 import java.io.Serializable;
-import java.util.List;
 
 /**
  * @author samuel
@@ -18,8 +17,6 @@ public class ShareCdcContext implements Serializable {
 	// Incremental start timestamp
 	private Long cdcStartTs;
 
-	private List<String> tableNames;
-
 	// Tapdata Settings
 	private ConfigurationCenter configurationCenter;
 
@@ -49,12 +46,4 @@ public TapLogger getObsLogger() {
 	public void setObsLogger(TapLogger obsLogger) {
 		this.obsLogger = obsLogger;
 	}
-
-	public List<String> getTableNames() {
-		return tableNames;
-	}
-
-	public void setTableNames(List<String> tableNames) {
-		this.tableNames = tableNames;
-	}
 }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNode.java
index 9bb5993357..5941e0518b 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNode.java
@@ -578,13 +578,12 @@ public void handleStreamEventsReceived(List<TapEvent> tapEvents, Object offsetOb
 //				}
 //			}
 			if (tapEvents != null && !tapEvents.isEmpty()) {
-				String taskSyncType = this.dataProcessorContext.getTaskDto().getSyncType();
 				tapEvents = tapEvents.stream().map(event -> {
 					if (null == event.getTime()) {
 						throw new NodeException("Invalid TapEvent, `TapEvent.time` should be NonNUll").context(getProcessorBaseContext()).event(event);
 					}
 					event.addInfo("eventId", UUID.randomUUID().toString());
-					return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)), taskSyncType);
+					return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)));
 				}).collect(Collectors.toList());
 
 				if (streamReadFuncAspect != null) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
index 44f5e8b251..f47e210fc5 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
@@ -983,7 +983,7 @@ protected List<TapdataEvent> wrapTapdataEvent(List<TapEvent> events, SyncStage s
 			if (null == tapEvent.getTime()) {
 				throw new NodeException("Invalid TapEvent, `TapEvent.time` should be NonNUll").context(getProcessorBaseContext()).event(tapEvent);
 			}
-			TapEvent tapEventCache = cdcDelayCalculation.filterAndCalcDelay(tapEvent, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)),this.dataProcessorContext.getTaskDto().getSyncType());
+			TapEvent tapEventCache = cdcDelayCalculation.filterAndCalcDelay(tapEvent, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)));
 			eventCache.add(tapEventCache);
 			boolean isLast = i == (size - 1);
 			TapdataEvent tapdataEvent;
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
index 4ded68a985..a6d39c08af 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
@@ -420,7 +420,7 @@ protected void doSnapshotInvoke(String tableName, DoSnapshotFunctions functions,
 													if (null == event.getTime()) {
 														throw new NodeException("Invalid TapEvent, `TapEvent.time` should be NonNUll").context(getProcessorBaseContext()).event(event);
 													}
-													return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)),this.dataProcessorContext.getTaskDto().getSyncType());
+													return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)));
 												}).collect(Collectors.toList());
 
 												if (batchReadFuncAspect != null)
@@ -812,7 +812,7 @@ protected StreamReadConsumer generateStreamReadConsumer(ConnectorNode connectorN
 						if (null == event.getTime()) {
 							throw new NodeException("Invalid TapEvent, `TapEvent.time` should be NonNUll").context(getProcessorBaseContext()).event(event);
 						}
-						return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)),this.dataProcessorContext.getTaskDto().getSyncType());
+						return cdcDelayCalculation.filterAndCalcDelay(event, times -> AspectUtils.executeAspect(SourceCDCDelayAspect.class, () -> new SourceCDCDelayAspect().delay(times).dataProcessorContext(dataProcessorContext)));
 					}).collect(Collectors.toList());
 
 					if (streamReadFuncAspect != null) {
@@ -912,15 +912,14 @@ protected void doShareCdc() throws Exception {
 		if (!isRunning()) {
 			return;
 		}
+		Optional.of(cdcDelayCalculation.addHeartbeatTable(new ArrayList<>(dataProcessorContext.getTapTableMap().keySet())))
+				.map(joinHeartbeat -> executeAspect(SourceJoinHeartbeatAspect.class, () -> new SourceJoinHeartbeatAspect().dataProcessorContext(dataProcessorContext).joinHeartbeat(joinHeartbeat)));
+		ShareCdcTaskContext shareCdcTaskContext = createShareCDCTaskContext();
 		TapTableMap<String, TapTable> tapTableMap = dataProcessorContext.getTapTableMap();
 		List<String> tables = new ArrayList<>(tapTableMap.keySet());
 		excludeRemoveTable(tables);
-		Optional.of(cdcDelayCalculation.addHeartbeatTable(tables))
-				.map(joinHeartbeat -> executeAspect(SourceJoinHeartbeatAspect.class, () -> new SourceJoinHeartbeatAspect().dataProcessorContext(dataProcessorContext).joinHeartbeat(joinHeartbeat)));
-		ShareCdcTaskContext shareCdcTaskContext = createShareCDCTaskContext();
 		this.syncProgressType = SyncProgress.Type.SHARE_CDC;
 		PDKMethodInvoker pdkMethodInvoker = createPdkMethodInvoker();
-		shareCdcTaskContext.setTableNames(tables);
 		try {
 			executeDataFuncAspect(StreamReadFuncAspect.class,
 					() -> new StreamReadFuncAspect()
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
index 8c6c48e6be..b9772a4936 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
@@ -761,6 +761,10 @@ protected ExternalStorageDto copyExternalStorage(int inMemSize) {
 		if (null == externalStorageDto) {
 			throw new TapCodeException(TaskMergeProcessorExCode_16.COPY_EXTERNAL_STORAGE_FAILED_SOURCE_IS_NULL);
 		}
+		return copyExternalStorage(externalStorageDto, inMemSize);
+	}
+
+	protected static ExternalStorageDto copyExternalStorage(ExternalStorageDto externalStorageDto,int inMemSize){
 		ExternalStorageDto externalStorageDtoCopy = new ExternalStorageDto();
 		BeanUtils.copyProperties(externalStorageDto, externalStorageDtoCopy);
 		externalStorageDtoCopy.setTable(null);
@@ -1668,13 +1672,15 @@ private io.tapdata.pdk.apis.entity.merge.MergeTableProperties copyMergeTableProp
 	public static void clearCache(Node<?> node) {
 		if (!(node instanceof MergeTableNode)) return;
 		ExternalStorageDto externalStorage = ExternalStorageUtil.getExternalStorage(node);
-		recursiveClearCache(externalStorage, ((MergeTableNode) node).getMergeProperties(), HazelcastUtil.getInstance());
+		ExternalStorageDto copyExternalStorage = copyExternalStorage(externalStorage,CommonUtils.getPropertyInt(MERGE_CACHE_IN_MEM_SIZE_PROP_KEY, DEFAULT_MERGE_CACHE_IN_MEM_SIZE));
+		recursiveClearCache(copyExternalStorage, ((MergeTableNode) node).getMergeProperties(), HazelcastUtil.getInstance());
 	}
 
 	public static void clearCache(Node<?> node, List<Node> nodes, List<Edge> edges) {
 		if (!(node instanceof MergeTableNode)) return;
 		ExternalStorageDto externalStorage = ExternalStorageUtil.getTargetNodeExternalStorage(node, edges, ConnectorConstant.clientMongoOperator, nodes);
-		recursiveClearCache(externalStorage, ((MergeTableNode) node).getMergeProperties(), HazelcastUtil.getInstance());
+		ExternalStorageDto copyExternalStorage = copyExternalStorage(externalStorage,CommonUtils.getPropertyInt(MERGE_CACHE_IN_MEM_SIZE_PROP_KEY, DEFAULT_MERGE_CACHE_IN_MEM_SIZE));
+		recursiveClearCache(copyExternalStorage, ((MergeTableNode) node).getMergeProperties(), HazelcastUtil.getInstance());
 	}
 
 	private static void recursiveClearCache(ExternalStorageDto externalStorageDto, List<MergeTableProperties> mergeTableProperties, HazelcastInstance hazelcastInstance) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskService.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskService.java
index a9a295d6e4..9b34367455 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskService.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskService.java
@@ -59,6 +59,7 @@
 import io.tapdata.error.TaskProcessorExCode_11;
 import io.tapdata.exception.NodeException;
 import io.tapdata.exception.TapCodeException;
+import io.tapdata.flow.engine.V2.cleaner.impl.MergeNodeCleaner;
 import io.tapdata.flow.engine.V2.entity.GlobalConstant;
 import io.tapdata.flow.engine.V2.log.LogFactory;
 import io.tapdata.flow.engine.V2.node.NodeTypeEnum;
@@ -345,6 +346,8 @@ protected JetDag task2HazelcastDAG(TaskDto taskDto,Boolean deduce) {
 						}
 					}
 					messageDao.registerCache((CacheNode) node, (TableNode) sourceNode, connection, taskDtoAtomicReference.get(), clientMongoOperator);
+				} else if (node instanceof MergeTableNode){
+					cleanMergeNode(taskDto,node.getId(),nodes);
 				}
 				List<Node> predecessors = node.predecessors();
 				List<Node> successors = node.successors();
@@ -984,4 +987,18 @@ protected Map<String,TapTableMap<String, TapTable>> engineTransformSchema(TaskDt
 		return tapTableMapHashMap;
 	}
 
+	protected void cleanMergeNode(TaskDto taskDto, String nodeId, List<Node> nodes){
+		AtomicBoolean containDisabledNode = new AtomicBoolean(false);
+		nodes.forEach(node -> {
+			if(node.disabledNode()){
+				containDisabledNode.set(true);
+			}
+		});
+		if(!containDisabledNode.get() && !taskDto.getType().equals(TaskDto.TYPE_CDC) && (null == taskDto.getAttrs() || !taskDto.getAttrs().containsKey("syncProgress"))){
+			MergeNodeCleaner mergeNodeCleaner = new MergeNodeCleaner();
+			mergeNodeCleaner.cleanTaskNode(taskDto.getId().toHexString(), nodeId);
+			logger.info("Clear {} master-slave merge cache", nodeId);
+		}
+	}
+
 }
diff --git a/iengine/iengine-app/src/main/resources/pluginKit.properties b/iengine/iengine-app/src/main/resources/pluginKit.properties
index a678dd401f..8250138336 100644
--- a/iengine/iengine-app/src/main/resources/pluginKit.properties
+++ b/iengine/iengine-app/src/main/resources/pluginKit.properties
@@ -12,7 +12,7 @@ maven.compiler.target=1.8
 project.build.sourceEncoding=UTF-8
 project.reporting.outputEncoding=UTF-8
 proxy.file.name=plugin-kit-v1.0-SNAPSHOT
-tapdata.api.verison=1.3.9-SNAPSHOT
-tapdata.pdk.api.verison=1.3.9-SNAPSHOT
+tapdata.api.verison=1.4.0-SNAPSHOT
+tapdata.pdk.api.verison=1.4.0-SNAPSHOT
 tapdata.pdk.connector.core.version=1.0-SNAPSHOT
 tapdata.pdk.runner.verison=1.9-SNAPSHOT
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/constant/DateUtil.java b/iengine/iengine-common/src/main/java/com/tapdata/constant/DateUtil.java
index d331d98fcd..6773913258 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/constant/DateUtil.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/constant/DateUtil.java
@@ -472,17 +472,11 @@ public static TimeZone getTimeZone(Integer timeZoneOffset) {
 			return TimeZone.getDefault();
 		}
 		StringBuilder sb = new StringBuilder("GMT");
-		String str = String.valueOf(timeZoneOffset);
-		if (str.length() <= 1 || (str.contains("-") && str.length() <= 2)) {
-			if (str.contains("-")) {
-				sb.append("-0").append(Math.abs(timeZoneOffset));
-			} else {
-				sb.append("+0").append(timeZoneOffset);
-			}
-		} else {
-			sb.append(str);
-		}
-		sb.append(":00");
+        if (timeZoneOffset < 0) {
+            sb.append(String.format("%02d:00", timeZoneOffset));
+        } else {
+            sb.append(String.format("+%02d:00", timeZoneOffset));
+        }
 
 		return TimeZone.getTimeZone(sb.toString());
 	}
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/constant/MongodbUtil.java b/iengine/iengine-common/src/main/java/com/tapdata/constant/MongodbUtil.java
index 8eb134e8f7..831ad3c5d5 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/constant/MongodbUtil.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/constant/MongodbUtil.java
@@ -71,6 +71,7 @@
 import org.bson.codecs.configuration.CodecRegistry;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
+import org.springframework.core.codec.EncodingException;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
@@ -122,6 +123,10 @@ public class MongodbUtil extends BaseDatabaseUtil {
 
 	private static final int SAMPLE_SIZE_BATCH_SIZE = 1000;
 
+	public final static String PREFIX_MONGO_URI = "mongodb://";
+
+	public final static String UTF_8 = "UTF-8";
+
 	public static final Set<String> systemTables = new HashSet<>();
 
 	static {
@@ -2251,4 +2256,35 @@ public static MongoClientURI verifyMongoDBUriWithDB(String uri) {
 			throw new IllegalArgumentException("MongoDB client uri missing database: " + maskUriPassword(uri));
 		return mongoClientURI;
 	}
+
+	public static String getUri(Map<String, Object> config) {
+		boolean isUri = MapUtils.getBooleanValue(config, "isUri");
+		if (isUri) {
+			return MapUtils.getString(config, "uri");
+		} else {
+			StringBuilder sb = new StringBuilder(PREFIX_MONGO_URI);
+			String user = MapUtils.getString(config, "user");
+			String password = MapUtils.getString(config, "password");
+			if (StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password)) {
+				String encodeUsername = null;
+				String encodePassword = null;
+				try {
+					encodeUsername = URLEncoder.encode(user, UTF_8);
+					encodePassword = URLEncoder.encode(password, UTF_8);
+				} catch (UnsupportedEncodingException e) {
+					throw new EncodingException(String.format("Encoding mongodb username/password failed %s", e.getMessage()), e);
+				}
+				sb.append(encodeUsername).append(":").append(encodePassword).append("@");
+			}
+			String host = MapUtils.getString(config, "host");
+			sb.append(host);
+			String database = MapUtils.getString(config, "database");
+			sb.append("/").append(database);
+			String additionalString = MapUtils.getString(config, "additionalString");
+			if (StringUtils.isNotBlank(additionalString)) {
+				sb.append("?").append(additionalString);
+			}
+			return sb.toString();
+		}
+	}
 }
diff --git a/iengine/iengine-common/src/main/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtil.java b/iengine/iengine-common/src/main/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtil.java
index 9090f60ee3..9dfff1c30a 100644
--- a/iengine/iengine-common/src/main/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtil.java
+++ b/iengine/iengine-common/src/main/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtil.java
@@ -315,14 +315,7 @@ public static ExternalStorageDto getTargetNodeExternalStorage(
 				Connections connection = clientMongoOperator.findOne(connQuery, ConnectorConstant.CONNECTION_COLLECTION, Connections.class);
 				if(connection.getDatabase_type().equals("MongoDB") || connection.getDatabase_type().equals("MongoDB Atlas")){
 					Map<String, Object> config =connection.getConfig();
-					String uri;
-					if((boolean)config.get("isUri")){
-						uri =(String)config.get("uri");
-					}else {
-						uri = "mongodb://%s:%s@%s/%s?%s";
-						uri = String.format(uri,config.get("user"),config.get("password"),
-								config.get("host"),config.get("database"),config.get("additionalString"));
-					}
+					String uri = MongodbUtil.getUri(config);
 					externalStorageDto.setName(ConnectorConstant.TARGET_MONGO_DB_EXTERNAL_STORAGE_NAME);
 					externalStorageDto.setUri(uri);
 					externalStorageDto.setType("mongodb");
@@ -339,6 +332,7 @@ public static ExternalStorageDto getTargetNodeExternalStorage(
 		return externalStorageDto;
 	}
 
+
 	public static ExternalStorageDto getExternalStorage(Node node) {
 		ClientMongoOperator clientMongoOperator = ConnectorConstant.clientMongoOperator;
 		return getExternalStorage(node, null, clientMongoOperator, null);
diff --git a/iengine/pom.xml b/iengine/pom.xml
index 572c8384ca..ea36639f58 100644
--- a/iengine/pom.xml
+++ b/iengine/pom.xml
@@ -799,7 +799,7 @@
             <dependency>
                 <groupId>com.tapdata</groupId>
                 <artifactId>tapdata-api</artifactId>
-                <version>1.3.9-SNAPSHOT</version>
+                <version>1.4.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.tapdata</groupId>
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java b/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
index aa804d8758..63ac5557c5 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
@@ -463,46 +463,8 @@ public static TestResponseDto sendHtmlEmail(MailAccountDto parms, List<String> a
             log.error("mail account info empty, params:{}", JSON.toJSONString(parms));
             return new TestResponseDto(false,"Please check your configuration, mail account information cannot be empty.");
         } else {
-            if (StringUtils.isNotBlank(parms.getProxyHost()) && 0 != parms.getProxyPort()) {
-                return sendEmailForProxy(parms, adressees, title, content, flag);
-            }
-            try {
-                MailAccount account = new MailAccount();
-                account.setHost(parms.getHost());
-                account.setPort(parms.getPort());
-                account.setAuth(true);
-                account.setFrom(parms.getFrom());
-                account.setUser(parms.getUser());
-                account.setPass(parms.getPass());
-                if ("SSL".equals(parms.getProtocol())) {
-                    // 使用SSL安全连接
-                    account.setSslEnable(true);
-                    //指定实现javax.net.SocketFactory接口的类的名称,这个类将被用于创建SMTP的套接字
-                    account.setSocketFactoryClass("javax.net.ssl.SSLSocketFactory");
-                } else if ("TLS".equals(parms.getProtocol())) {
-                    account.setStarttlsEnable(true);
-                    account.setSocketFactoryClass("javax.net.ssl.SSLSocketFactory");
-                } else {
-                    account.setSslEnable(false);
-                    account.setStarttlsEnable(false);
-                }
-
-                //如果设置为true,未能创建一个套接字使用指定9的套接字工厂类将导致使用java.net.Socket创建的套接字类, 默认值为true
-                account.setSocketFactoryFallback(true);
-                // 指定的端口连接到在使用指定的套接字工厂。如果没有设置,将使用默认端口456
-                account.setSocketFactoryPort(465);
-                Map<String, Object> oemConfig = OEMReplaceUtil.getOEMConfigMap("email/replace.json");
-                title = OEMReplaceUtil.replace(title, oemConfig);
-                content = OEMReplaceUtil.replace(assemblyMessageBody(content), oemConfig);
-                MailUtil.send(account, adressees, title ,content, true);
-            } catch (Exception e) {
-                log.error("mail send error：{}", e.getMessage(), e);
-                flag = false;
-                return new TestResponseDto(flag, TapSimplify.getStackTrace(e));
-            }
+            return sendEmailForProxy(parms, adressees, title, content, flag);
         }
-        log.debug("mail send status：{}", flag ? "suc" : "error");
-        return new TestResponseDto(flag, null);
     }
 
     @Nullable
@@ -543,9 +505,10 @@ protected static TestResponseDto sendEmailForProxy(MailAccountDto parms, List<St
             properties.put("mail.smtp.starttls.enable", "false");
         }
         //set proxy server
-        properties.put("mail.smtp.socks.host", parms.getProxyHost());
-        properties.put("mail.smtp.socks.port", parms.getProxyPort());
-
+        if (StringUtils.isNotBlank(parms.getProxyHost()) && 0 != parms.getProxyPort()){
+            properties.put("mail.smtp.socks.host", parms.getProxyHost());
+            properties.put("mail.smtp.socks.port", parms.getProxyPort());
+        }
         Session session = Session.getInstance(properties, new Authenticator() {
             protected PasswordAuthentication getPasswordAuthentication() {
                 return new PasswordAuthentication(username, password);
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
index 97d0d873af..0a0f14ffce 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
@@ -52,13 +52,13 @@ public boolean addHeartbeatTable(@NonNull List<String> tables) {
     }
 
     @Override
-    public TapEvent filterAndCalcDelay(TapEvent tapEvent, @NonNull LongConsumer delayConsumer, String syncType) {
+    public TapEvent filterAndCalcDelay(TapEvent tapEvent, @NonNull LongConsumer delayConsumer) {
         if (tapEvent instanceof TapRecordEvent) {
             Long sourceTimes;
             TapRecordEvent tapRecordEvent = ((TapRecordEvent) tapEvent);
             if (ConnHeartbeatUtils.TABLE_NAME.equals(tapRecordEvent.getTableId())) {
                 sourceTimes = parseTs(tapRecordEvent);
-                if (isFilter && !TaskDto.SYNC_TYPE_LOG_COLLECTOR.equals(syncType)) {
+                if (isFilter) {
                     HeartbeatEvent heartbeatEvent = new HeartbeatEvent();
                     tapEvent.clone(heartbeatEvent);
                     heartbeatEvent.setReferenceTime(tapRecordEvent.getReferenceTime());
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelayDisable.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelayDisable.java
index 12c5cb2a29..460b3900f7 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelayDisable.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelayDisable.java
@@ -17,7 +17,7 @@ public boolean addHeartbeatTable(@NonNull List<String> tables) {
     }
 
     @Override
-    public TapEvent filterAndCalcDelay(TapEvent tapEvent, LongConsumer delayConsumer, String syncType) {
+    public TapEvent filterAndCalcDelay(TapEvent tapEvent, LongConsumer delayConsumer) {
         return tapEvent;
     }
 }
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/ICdcDelay.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/ICdcDelay.java
index e6f92bc02c..7f7e6d2eee 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/ICdcDelay.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/ICdcDelay.java
@@ -25,5 +25,5 @@ public interface ICdcDelay {
      * @param tapEvent TapEvent
      * @return TapEvent
      */
-    TapEvent filterAndCalcDelay(TapEvent tapEvent, @NonNull LongConsumer delayConsumer, String taskSyncType);
+    TapEvent filterAndCalcDelay(TapEvent tapEvent, @NonNull LongConsumer delayConsumer);
 }
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/ConnHeartbeatUtils.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/ConnHeartbeatUtils.java
index c3451f9818..dab39e5986 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/ConnHeartbeatUtils.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/ConnHeartbeatUtils.java
@@ -23,10 +23,14 @@ public class ConnHeartbeatUtils {
     public static final String PDK_ID = "dummy";
     public static final String PDK_NAME = "Dummy";
     public static final String MODE = "ConnHeartbeat";
+    public static final String CONNECTOR_TAGS_NO_HEARTBEAT = "NoHeartbeat";
     public static final String CONNECTION_NAME = "tapdata_heartbeat_dummy_connection";
     public static final String TABLE_NAME = "_tapdata_heartbeat_table";
     public static final String TASK_RELATION_FIELD = "heartbeatTasks";
 
+    private ConnHeartbeatUtils() {
+    }
+
     /**
      * check the task need to start heartbeat task
      *
@@ -40,6 +44,12 @@ public static boolean checkTask(@NonNull String taskType, @NonNull String taskSy
         ;
     }
 
+    public static boolean hasNoHeartbeatTag(DataSourceConnectionDto connectionDto) {
+        List<String> tags = connectionDto.getDefinitionTags();
+        if (null == tags || tags.isEmpty()) return false;
+        return tags.contains(CONNECTOR_TAGS_NO_HEARTBEAT);
+    }
+
     /**
      * check the connection can be start heartbeat task
      *
@@ -47,17 +57,19 @@ public static boolean checkTask(@NonNull String taskType, @NonNull String taskSy
      * @return can start heartbeat
      */
     public static boolean checkConnection(@NonNull DataSourceConnectionDto sourceConnectionDto) {
+        List<Capability> capabilities = sourceConnectionDto.getCapabilities();
         if (!Boolean.TRUE.equals(sourceConnectionDto.getHeartbeatEnable())
                 || PDK_NAME.equals(sourceConnectionDto.getDatabase_type())
-                || null == sourceConnectionDto.getCapabilities()
+                || null == capabilities
                 || "source".equalsIgnoreCase(sourceConnectionDto.getConnection_type())
                 || "target".equalsIgnoreCase(sourceConnectionDto.getConnection_type())
+                || hasNoHeartbeatTag(sourceConnectionDto)
         ) {
             return false;
         }
 
         boolean hasStreamRead = false, hasCreateTable = false, hasWriteRecord = false;
-        for (Capability capability : sourceConnectionDto.getCapabilities()) {
+        for (Capability capability : capabilities) {
             if (CapabilityEnum.STREAM_READ_FUNCTION.name().equalsIgnoreCase(capability.getId())) {
                 hasStreamRead = true;
             } else if (CapabilityEnum.CREATE_TABLE_FUNCTION.name().equalsIgnoreCase(capability.getId())
diff --git a/manager/tm-sdk/pom.xml b/manager/tm-sdk/pom.xml
index 8b8bce63a6..d9669de7db 100644
--- a/manager/tm-sdk/pom.xml
+++ b/manager/tm-sdk/pom.xml
@@ -22,7 +22,7 @@
 		<dependency>
 			<groupId>io.tapdata</groupId>
 			<artifactId>tapdata-api</artifactId>
-			<version>1.3.9-SNAPSHOT</version>
+			<version>1.4.0-SNAPSHOT</version>
 		</dependency>
 
     <dependency>
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..b0053f9e5d 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,22 @@ 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
+				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 +148,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 +156,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);
 			}
@@ -158,27 +180,15 @@ public void uploadPdk(CommonsMultipartFile[] files, List<PdkSourceDto> pdkSource
 				Update removeLatest = Update.update("latest", false);
 				dataSourceDefinitionService.update(new Query(criteriaLatest), removeLatest);
 			}
+			definitionDto.setId(null);
 			if (Objects.isNull(oldDefinitionDto)) {
 				dataSourceDefinitionService.save(definitionDto, user);
 			} else {
-				dataSourceDefinitionService.upsert(Query.query(Criteria.where("_id").is(definitionDto.getId())), definitionDto, user);
+				dataSourceDefinitionService.upsert(Query.query(Criteria.where("_id").is(oldDefinitionDto.getId())), definitionDto, user);
 			}
-
-			//根据数据源类型删除可能存在的旧的pdk
-			FunctionUtils.ignoreAnyError(() -> {
-				Object buildProfile = settingsService.getByCategoryAndKey("System", "buildProfile");
-				if (Objects.isNull(buildProfile)) {
-					buildProfile = "DAAS";
-				}
-
-				boolean isCloud = buildProfile.equals("CLOUD") || buildProfile.equals("DRS") || buildProfile.equals("DFS");
-				if (!isCloud) {
-					Query query = DataSourceDefinitionServiceImpl.getQueryByDatasourceType(Lists.of(definitionDto.getType()), user, definitionDto.getId());
-					dataSourceDefinitionService.deleteAll(query);
-				}
-			});
-
+			log.debug("Upsert data source definition success");
 		}
+		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/init/patches/daas/V3_14_3_ScheduleTasks_status_pingTime_Index.java b/manager/tm/src/main/java/com/tapdata/tm/init/patches/daas/V3_14_3_ScheduleTasks_status_pingTime_Index.java
new file mode 100644
index 0000000000..0529d8361c
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/init/patches/daas/V3_14_3_ScheduleTasks_status_pingTime_Index.java
@@ -0,0 +1,33 @@
+package com.tapdata.tm.init.patches.daas;
+
+import com.tapdata.tm.init.PatchType;
+import com.tapdata.tm.init.PatchVersion;
+import com.tapdata.tm.init.patches.AbsPatch;
+import com.tapdata.tm.init.patches.PatchAnnotation;
+import com.tapdata.tm.utils.SpringContextHelper;
+import io.tapdata.utils.AppType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.index.Index;
+
+@PatchAnnotation(appType = AppType.DAAS, version = "3.14-3")
+public class V3_14_3_ScheduleTasks_status_pingTime_Index extends AbsPatch {
+    private static final Logger logger = LogManager.getLogger(V3_14_3_ScheduleTasks_status_pingTime_Index.class);
+
+    public V3_14_3_ScheduleTasks_status_pingTime_Index(PatchType type, PatchVersion version) {
+        super(type, version);
+    }
+
+    @Override
+    public void run() {
+        logger.info("Execute java patch: {}...", getClass().getName());
+        String collectionName = "ScheduleTasks";
+        MongoTemplate mongoTemplate = SpringContextHelper.getBean(MongoTemplate.class);
+
+        Index index = new Index().named("status_1_ping_time_1").on("status", Sort.Direction.ASC).on("ping_time", Sort.Direction.ASC).background();
+        mongoTemplate.indexOps(collectionName).ensureIndex(index);
+    }
+
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_14_3_ScheduleTasks_status_pingTime_Index.java b/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_14_3_ScheduleTasks_status_pingTime_Index.java
new file mode 100644
index 0000000000..19041fe950
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_14_3_ScheduleTasks_status_pingTime_Index.java
@@ -0,0 +1,33 @@
+package com.tapdata.tm.init.patches.dfs;
+
+import com.tapdata.tm.init.PatchType;
+import com.tapdata.tm.init.PatchVersion;
+import com.tapdata.tm.init.patches.AbsPatch;
+import com.tapdata.tm.init.patches.PatchAnnotation;
+import com.tapdata.tm.utils.SpringContextHelper;
+import io.tapdata.utils.AppType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.index.Index;
+
+@PatchAnnotation(appType = AppType.DFS, version = "3.14-3")
+public class V3_14_3_ScheduleTasks_status_pingTime_Index extends AbsPatch {
+    private static final Logger logger = LogManager.getLogger(V3_14_3_ScheduleTasks_status_pingTime_Index.class);
+
+    public V3_14_3_ScheduleTasks_status_pingTime_Index(PatchType type, PatchVersion version) {
+        super(type, version);
+    }
+
+    @Override
+    public void run() {
+        logger.info("Execute java patch: {}...", getClass().getName());
+        String collectionName = "ScheduleTasks";
+        MongoTemplate mongoTemplate = SpringContextHelper.getBean(MongoTemplate.class);
+
+        Index index = new Index().named("status_1_ping_time_1").on("status", Sort.Direction.ASC).on("ping_time", Sort.Direction.ASC).background();
+        mongoTemplate.indexOps(collectionName).ensureIndex(index);
+    }
+
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImpl.java
index b82fdc4f49..620b3728ff 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImpl.java
@@ -1488,10 +1488,7 @@ public String findHeartbeatQualifiedNameByNodeId(Filter filter, UserDetail user)
             return taskService.findOne(query, user);
         }).map(task -> task.getId().toHexString()).map(tid -> {
             // get heartbeat task dag of the connection node
-            Criteria criteria = Criteria.where(ConnHeartbeatUtils.TASK_RELATION_FIELD).is(tid);
-            criteria.and("status").ne("deleting");
-            criteria.and(IS_DELETED).ne(true);
-            Query query = new Query(criteria);
+            Query query = new Query(Criteria.where(ConnHeartbeatUtils.TASK_RELATION_FIELD).is(tid));
             query.fields().include("_id", "dag");
             return taskService.findOne(query, user);
         }).map(taskDto -> {
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/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
index 4b64298285..cd4947efb4 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
@@ -4038,7 +4038,6 @@ public void start(TaskDto taskDto, UserDetail user, String startFlag) {
             log.warn("heartbeat task current status not allow to start, task = {}, status = {}, please restore the task manually.", taskDto.getName(), taskDto.getStatus());
             return;
         }
-        checkHeartTableInLogCollectorTaskTable(taskDto, user);
         //校验当前状态是否允许启动。
         if (!TaskOpStatusEnum.to_start_status.v().contains(taskDto.getStatus())) {
             log.warn("task current status not allow to start, task = {}, status = {}", taskDto.getName(), taskDto.getStatus());
@@ -4072,46 +4071,6 @@ public void start(TaskDto taskDto, UserDetail user, String startFlag) {
         }
     }
 
-    protected void checkHeartTableInLogCollectorTaskTable(TaskDto taskDto, UserDetail user) {
-        if (TaskDto.SYNC_TYPE_LOG_COLLECTOR.equals(taskDto.getSyncType())) {
-            Node logCollectorTaskSourceNode = taskDto.getDag().getSourceNodes().get(0);
-            if ((logCollectorTaskSourceNode instanceof LogCollectorNode)) {
-                LogCollectorNode logCollectorNode = (LogCollectorNode) logCollectorTaskSourceNode;
-                List<String> connectionIds = logCollectorNode.getConnectionIds();
-                Set<String> tableNameSet = new HashSet<>(logCollectorNode.getTableNames());
-                Map<String, LogCollecotrConnConfig> logCollectorConnConfigs = logCollectorNode.getLogCollectorConnConfigs();
-                List<DataSourceConnectionDto> allDataSourceConnection = dataSourceService.findAllByIds(connectionIds);
-                boolean hasUpdate = addHeartBeatTable2LogCollector(allDataSourceConnection, logCollectorConnConfigs, tableNameSet);
-                if (hasUpdate) {
-                    List<String> finalTableNames = new ArrayList<>(tableNameSet);
-                    logCollectorNode.setTableNames(finalTableNames);
-                    shareCdcTableMappingService.genShareCdcTableMappingsByLogCollectorTask(taskDto, false, user);
-                    updateById(taskDto, user);
-                }
-            }
-        }
-    }
-
-    protected boolean addHeartBeatTable2LogCollector(List<DataSourceConnectionDto> allDataSourceConnection, Map<String, LogCollecotrConnConfig> logCollectorConnConfigs, Set<String> tableNameSet) {
-        boolean updateConfig = false;
-        int beforeTableNamesSize = tableNameSet.size();
-        for (DataSourceConnectionDto dataSourceConnectionDto : allDataSourceConnection) {
-            if (Boolean.TRUE.equals(dataSourceConnectionDto.getHeartbeatEnable())) {
-                if (MapUtils.isNotEmpty(logCollectorConnConfigs) && null != logCollectorConnConfigs.get(dataSourceConnectionDto.getId().toHexString())) {
-                    LogCollecotrConnConfig logCollecotrConnConfig = logCollectorConnConfigs.get(dataSourceConnectionDto.getId().toHexString());
-                    List<String> tableNames = logCollecotrConnConfig.getTableNames();
-                    if (tableNames.stream().noneMatch(ConnHeartbeatUtils.TABLE_NAME::equals)) {
-                        logCollecotrConnConfig.getTableNames().add(ConnHeartbeatUtils.TABLE_NAME);
-                        updateConfig = true;
-                    }
-                }
-                tableNameSet.add(ConnHeartbeatUtils.TABLE_NAME);
-            }
-        }
-        return updateConfig || beforeTableNamesSize < tableNameSet.size();
-    }
-
-
     /**
      * @see com.tapdata.tm.statemachine.enums.DataFlowEvent#START
      * @param taskDto
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
index 1c7dcdd8a8..2acfad3171 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
@@ -1078,7 +1078,7 @@ public Map<String, String> ldpTableStatus(String connectionId, List<String> tabl
                         if (ldpNewTables == null) {
                             ldpNewTables = new ArrayList<>();
                         }
-                        if (TaskDto.STATUS_RUNNING.equals(taskDto.getStatus())) {
+                        if (TaskDto.STATUS_RUNNING.equals(taskDto.getStatus()) || TaskDto.STATUS_COMPLETE.equals(taskDto.getStatus())) {
                             String state = "running";
                             for (String tableName : tableNames) {
                                 if (map.containsKey(tableName)) {
@@ -1124,7 +1124,7 @@ public Map<String, String> ldpTableStatus(String connectionId, List<String> tabl
                         if (target instanceof TableNode) {
                             String tableName = ((TableNode) target).getTableName();
                             if (tableNames.contains(tableName)) {
-                                if (TaskDto.STATUS_RUNNING.equals(taskDto.getStatus())) {
+                                if (TaskDto.STATUS_RUNNING.equals(taskDto.getStatus()) || TaskDto.STATUS_COMPLETE.equals(taskDto.getStatus())) {
                                     tableStatusMap.put(tableName, "running");
                                 } else {
                                     if (!"running".equals(tableStatusMap.get(tableName))) {
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskDagServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskDagServiceImpl.java
index 77a78f072c..b22fca036b 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskDagServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskDagServiceImpl.java
@@ -28,6 +28,8 @@
 import java.io.IOException;
 import java.util.*;
 
+import static org.apache.commons.lang3.RandomUtils.nextInt;
+
 /**
  * @author jiuyetx
  * @date 2022/8/30
@@ -35,10 +37,11 @@
 @Service
 @Slf4j
 public class TaskDagServiceImpl implements TaskDagService {
+    private Random random = new Random();
     @Override
     public int calculationDagHash(TaskDto taskDto) {
         if (true) {
-            return new Random().nextInt();
+            return random.nextInt();
         }
         DAG dag = taskDto.getDag();
 
diff --git a/manager/tm/src/main/java/com/tapdata/tm/worker/service/WorkerServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/worker/service/WorkerServiceImpl.java
index e80a0da023..22dbe40f2a 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/worker/service/WorkerServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/worker/service/WorkerServiceImpl.java
@@ -100,6 +100,7 @@ public class WorkerServiceImpl extends WorkerService{
     private MongoTemplate mongoTemplate;
 
     private final MultiTaggedCounter workerPing;
+    private Random random = new Random();
 
     public WorkerServiceImpl(@NonNull WorkerRepository repository) {
         super(repository);
@@ -732,7 +733,7 @@ public void createShareWorker(WorkerExpireDto workerExpireDto, UserDetail loginU
         String[] shareAgentUser = shareAgentUserValue.toString().split(",");
 
         // get share agent user by random
-        String shareAgentUserRandom = shareAgentUser[new Random().nextInt(shareAgentUser.length)];
+        String shareAgentUserRandom = shareAgentUser[random.nextInt(shareAgentUser.length)];
 
         WorkerExpire workerExpire = new WorkerExpire();
         workerExpire.setUserId(loginUser.getUserId());
diff --git a/manager/tm/src/main/resources/init/dfs/version b/manager/tm/src/main/resources/init/dfs/version
index 2abbcb3621..1a1ace7dd1 100644
--- a/manager/tm/src/main/resources/init/dfs/version
+++ b/manager/tm/src/main/resources/init/dfs/version
@@ -1,1 +1,1 @@
-3.12-1
\ No newline at end of file
+3.14-3
\ No newline at end of file
diff --git a/manager/tm/src/main/resources/init/idaas/version b/manager/tm/src/main/resources/init/idaas/version
index fcf88bcc8c..1a1ace7dd1 100644
--- a/manager/tm/src/main/resources/init/idaas/version
+++ b/manager/tm/src/main/resources/init/idaas/version
@@ -1,1 +1,1 @@
-3.14-2
\ No newline at end of file
+3.14-3
\ No newline at end of file
diff --git a/manager/tm/src/main/resources/messages.properties b/manager/tm/src/main/resources/messages.properties
index 64ed4559cb..a6c73d4fb0 100644
--- a/manager/tm/src/main/resources/messages.properties
+++ b/manager/tm/src/main/resources/messages.properties
@@ -53,6 +53,7 @@ License.NodeInstanceIdInvalid=Please make sure the node [{0}] is datasource node
 Task.ResetAgentNotFound=The reset task did not find an available engine, please start/restart the engine and try again
 Task.ResetStatusInvalid=The current status of the task that needs to be reset has been updated, please refresh the page and try again
 Task.NotfoundDatasource=Not found datasource '{0}'
+Task.HeartbeatStartError=The heartbeat task is abnormal. Please turn off the heartbeat option of the source node connection ({0}) or try to restart the current task.
 LogCollect.ConfigUpdateError=Current conditions do not allow modification
 LogCollect.ExternalStorageUpdateError=Current conditions do not allow to modify external storage
 LogCollect.UriInvalid=Mongodb uri filling is illegal
diff --git a/manager/tm/src/main/resources/messages_zh_CN.properties b/manager/tm/src/main/resources/messages_zh_CN.properties
index fa000c42f4..3e6662ac73 100644
--- a/manager/tm/src/main/resources/messages_zh_CN.properties
+++ b/manager/tm/src/main/resources/messages_zh_CN.properties
@@ -77,6 +77,7 @@ License.NodeInstanceIdInvalid=\u8BF7\u786E\u4FDD\u8282\u70B9 [{0}] \u4E3A\u6570\
 Task.ResetAgentNotFound=\u91CD\u7F6E\u4EFB\u52A1\u6CA1\u6709\u627E\u5230\u53EF\u7528\u7684\u5F15\u64CE\uFF0C\u8BF7\u542F\u52A8/\u91CD\u542F\u5F15\u64CE\u540E\u518D\u8BD5
 Task.ResetStatusInvalid=\u5F53\u524D\u9700\u8981\u91CD\u7F6E\u7684\u4EFB\u52A1\u72B6\u6001\u88AB\u66F4\u65B0\uFF0C\u8BF7\u5237\u65B0\u9875\u9762\u518D\u8BD5
 Task.NotfoundDatasource=\u627E\u4E0D\u5230\u6570\u636E\u6E90 '{0}'
+Task.HeartbeatStartError=\u5FC3\u8DF3\u4EFB\u52A1\u5F02\u5E38\uFF0C\u8BF7\u5173\u95ED\u6E90\u8282\u70B9\u8FDE\u63A5\uFF08{0}\uFF09\u7684\u5FC3\u8DF3\u9009\u9879\u6216\u5C1D\u8BD5\u91CD\u542F\u52A8\u5F53\u524D\u4EFB\u52A1
 LogCollect.ConfigUpdateError=\u5F53\u524D\u6761\u4EF6\u4E0D\u5141\u8BB8\u4FEE\u6539
 LogCollect.ExternalStorageUpdateError=\u5F53\u524D\u6761\u4EF6\u4E0D\u5141\u8BB8\u4FEE\u6539\u5916\u5B58\u914D\u7F6E
 LogCollect.UriInvalid=Mongodb uri \u586B\u5199\u4E0D\u5408\u6CD5
@@ -307,4 +308,4 @@ inspect.result.not.exists=\u5F53\u524D\u6821\u9A8C\u4EFB\u52A1\u7684\u6821\u9A8C
 
 source.setting.check.cdc= \u589E\u91CF\u68C0\u67E5\u3011\u0020\u6570\u636E\u6E90:{0}\u7684\u589E\u91CF\u529F\u80FD\u4E0D\u652F\u6301\uFF0C\u4E0D\u80FD\u914D\u7F6E\u589E\u91CF\u4EFB\u52A1\u000d\u000a
 
-export.inspectDetails.failed =\u4E0B\u8F7D\u6821\u9A8C\u8BE6\u60C5\u5931\u8D25 :{0}
\ No newline at end of file
+export.inspectDetails.failed =\u4E0B\u8F7D\u6821\u9A8C\u8BE6\u60C5\u5931\u8D25 :{0}
diff --git a/tapdata-cli/src/main/java/io/tapdata/pdk/cli/RegisterMain.java b/tapdata-cli/src/main/java/io/tapdata/pdk/cli/RegisterMain.java
index eec215226c..c63156f229 100644
--- a/tapdata-cli/src/main/java/io/tapdata/pdk/cli/RegisterMain.java
+++ b/tapdata-cli/src/main/java/io/tapdata/pdk/cli/RegisterMain.java
@@ -83,6 +83,7 @@ private enum ConnectorEnums {
         HUDI(BETA),
         JSON(ALPHA, TAG_FILE),
         KAFKA(GA, TAG_MQ),
+        KAFKA_ENHANCED(GA, TAG_MQ),
         LARK_APPROVAL(BETA),
         LARK_BITABLE(BETA),
         LARK_DOC(BETA),
@@ -110,7 +111,6 @@ private enum ConnectorEnums {
         SALESFORCE(ALPHA),
         SELECTDB(BETA),
         SHEIN(BETA),
-        STD_KAFKA(BETA, TAG_MQ),
         TABLESTORE(ALPHA),
         TDENGINE(BETA),
         TEMU(ALPHA),
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/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNodeTest.java
index c167eca7c5..e21f6cdee4 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePartitionReadDataNodeTest.java
@@ -58,9 +58,9 @@ void test1() {
         HeartbeatEvent heartbeatEvent2 = new HeartbeatEvent();
         TapEvent tapEvent = tapUpdateRecordEvent;
         tapEvent.clone(heartbeatEvent2);
-        when(cdcDelay.filterAndCalcDelay(tapEvent, null, null)).thenReturn(heartbeatEvent2);
+        when(cdcDelay.filterAndCalcDelay(tapEvent, null)).thenReturn(heartbeatEvent2);
         hazelcastSourcePartitionReadDataNode.handleStreamEventsReceived(tapEvents, null);
-        verify(cdcDelay, times(1)).filterAndCalcDelay(any(), any(), any());
+        verify(cdcDelay, times(1)).filterAndCalcDelay(any(), any());
     }
     @DisplayName("test streamReadFuncAspect is null ")
     @Test
@@ -80,9 +80,9 @@ void test2(){
         HeartbeatEvent hbEvent = new HeartbeatEvent();
         TapEvent tapEvent = tapUpdateRecordEvent;
         tapEvent.clone(hbEvent);
-        when(cdcDelay.filterAndCalcDelay(tapEvent, null, null)).thenReturn(hbEvent);
+        when(cdcDelay.filterAndCalcDelay(tapEvent, null)).thenReturn(hbEvent);
         hazelcastSourcePartitionReadDataNode.handleStreamEventsReceived(tapEvents, null);
-        verify(cdcDelay, times(1)).filterAndCalcDelay(any(), any(), any());
+        verify(cdcDelay, times(1)).filterAndCalcDelay(any(), any());
     }
 
 
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
index 28430aa337..ecd7777457 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
@@ -1736,7 +1736,7 @@ void test1() throws InterruptedException {
 			doCallRealMethod().when(hazelcastSourcePdkDataNode).generateStreamReadConsumer(any(),any());
 			StreamReadConsumer streamReadConsumer = hazelcastSourcePdkDataNode.generateStreamReadConsumer(connectorNode,pdkMethodInvoker);
 			streamReadConsumer.accept(tapEvents,null);
-			verify(cdcDelay,times(1)).filterAndCalcDelay(any(),any(),any());
+			verify(cdcDelay,times(1)).filterAndCalcDelay(any(),any());
 		}
 	}
 
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
index dde88c8153..9f244bef0c 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
@@ -7,8 +7,10 @@
 import com.hazelcast.map.IMap;
 import com.hazelcast.persistence.ConstructType;
 import com.hazelcast.persistence.PersistenceStorage;
+import com.tapdata.constant.HazelcastUtil;
 import com.tapdata.entity.TapdataHeartbeatEvent;
 import com.tapdata.entity.task.context.ProcessorBaseContext;
+import com.tapdata.tm.commons.dag.Edge;
 import io.tapdata.entity.schema.value.TapArrayValue;
 import io.tapdata.entity.schema.value.TapMapValue;
 import io.tapdata.utils.AppType;
@@ -2422,4 +2424,49 @@ void test_oldConstructIMapIsEmpty(){
 		}
 
 	}
+	@Nested
+	class CopyExternalStorageTest{
+		@Test
+		void test(){
+			ExternalStorageDto externalStorageDto = new ExternalStorageDto();
+			ExternalStorageDto result = HazelcastMergeNode.copyExternalStorage(externalStorageDto,1);
+			Assertions.assertNull(result.getTable());
+			Assertions.assertEquals(1,result.getInMemSize());
+			Assertions.assertEquals(10,result.getWriteDelaySeconds());
+			Assertions.assertEquals(0,result.getTtlDay());
+		}
+
+	}
+	@Nested
+	class ClearCacheTest{
+		@Test
+		void testDass(){
+			try(MockedStatic<ExternalStorageUtil> externalStorageUtil = mockStatic(ExternalStorageUtil.class);
+				MockedStatic<HazelcastUtil> hazelcastUtil = mockStatic(HazelcastUtil.class)){
+				externalStorageUtil.when(()->ExternalStorageUtil.getExternalStorage(any())).thenAnswer(invocation -> {
+					Node<?> node = invocation.getArgument(0);
+					Assertions.assertEquals(mergeTableNode,node);
+					return new ExternalStorageDto();
+				});
+				hazelcastUtil.when(()->HazelcastUtil.getInstance()).thenReturn(mock(HazelcastInstance.class));
+				HazelcastMergeNode.clearCache(mergeTableNode);
+			}
+		}
+
+		@Test
+		void testCloud(){
+			try(MockedStatic<ExternalStorageUtil> externalStorageUtil = mockStatic(ExternalStorageUtil.class);
+				MockedStatic<HazelcastUtil> hazelcastUtil = mockStatic(HazelcastUtil.class)){
+				List<Node> nodes = new ArrayList<>();
+				List<Edge> edges = new ArrayList<>();
+				externalStorageUtil.when(()->ExternalStorageUtil.getTargetNodeExternalStorage(any(),any(),any(),anyList())).thenAnswer(invocation -> {
+					Node<?> node = invocation.getArgument(0);
+					Assertions.assertEquals(mergeTableNode,node);
+					return new ExternalStorageDto();
+				});
+				hazelcastUtil.when(()->HazelcastUtil.getInstance()).thenReturn(mock(HazelcastInstance.class));
+				HazelcastMergeNode.clearCache(mergeTableNode,nodes,edges);
+			}
+		}
+	}
 }
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskServiceTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskServiceTest.java
index 657545f9b0..f9fb3917c7 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskServiceTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskServiceTest.java
@@ -2,12 +2,10 @@
 
 import com.hazelcast.core.HazelcastInstance;
 import com.mongodb.MongoClientException;
-import com.tapdata.constant.ConfigurationCenter;
-import com.tapdata.constant.ConnectionUtil;
-import com.tapdata.constant.ConnectorConstant;
-import com.tapdata.constant.HazelcastUtil;
+import com.tapdata.constant.*;
 import com.tapdata.entity.Connections;
 import com.tapdata.entity.DatabaseTypeEnum;
+import com.tapdata.entity.dataflow.SyncProgress;
 import com.tapdata.entity.task.config.TaskConfig;
 import com.tapdata.entity.task.config.TaskRetryConfig;
 import com.tapdata.mongo.ClientMongoOperator;
@@ -843,4 +841,97 @@ void testNormalTaskTapTableMapHashMapIsNotNull(){
             Assertions.assertEquals(except,result);
         }
     }
+
+    @Nested
+    class CleanMergeNodeTest{
+        HazelcastTaskService hazelcastTaskService = mock(HazelcastTaskService.class);
+        @Test
+        void test_cleanMergeNode(){
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+            taskDto.setType(TaskDto.TYPE_INITIAL_SYNC_CDC);
+            taskDto.setAttrs(new HashMap<>());
+            List<Node> nodes = new ArrayList<>();
+            MergeTableNode node = new MergeTableNode();
+            Map<String, Object> attrs = new HashMap<>();
+            attrs.put("disabled",false);
+            node.setAttrs(attrs);
+            nodes.add(node);
+            try (MockedStatic<BeanUtil> beanUtilMockedStatic = mockStatic(BeanUtil.class)) {
+                ClientMongoOperator clientMongoOperator = mock(ClientMongoOperator.class);
+                when(clientMongoOperator.findOne(any(Query.class), anyString(), eq(TaskDto.class))).thenReturn(null);
+                beanUtilMockedStatic.when(() -> BeanUtil.getBean(ClientMongoOperator.class)).thenReturn(clientMongoOperator);
+                doCallRealMethod().when(hazelcastTaskService).cleanMergeNode(taskDto,"test",nodes);
+                hazelcastTaskService.cleanMergeNode(taskDto,"test",nodes);
+                verify(clientMongoOperator, times(1)).findOne(any(Query.class), anyString(), eq(TaskDto.class));
+            }
+        }
+
+        @Test
+        void test_ContainDisabledNode(){
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+            taskDto.setType(TaskDto.TYPE_INITIAL_SYNC_CDC);
+            List<Node> nodes = new ArrayList<>();
+            MergeTableNode node = new MergeTableNode();
+            Map<String, Object> attrs = new HashMap<>();
+            attrs.put("disabled",true);
+            node.setAttrs(attrs);
+            nodes.add(node);
+            try (MockedStatic<BeanUtil> beanUtilMockedStatic = mockStatic(BeanUtil.class)) {
+                ClientMongoOperator clientMongoOperator = mock(ClientMongoOperator.class);
+                when(clientMongoOperator.findOne(any(Query.class), anyString(), eq(TaskDto.class))).thenReturn(null);
+                beanUtilMockedStatic.when(() -> BeanUtil.getBean(ClientMongoOperator.class)).thenReturn(clientMongoOperator);
+                doCallRealMethod().when(hazelcastTaskService).cleanMergeNode(taskDto,"test",nodes);
+                hazelcastTaskService.cleanMergeNode(taskDto,"test",nodes);
+                verify(clientMongoOperator, times(0)).findOne(any(Query.class), anyString(), eq(TaskDto.class));
+            }
+        }
+
+        @Test
+        void test_TaskTypeCDC(){
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+            taskDto.setType(TaskDto.TYPE_CDC);
+            List<Node> nodes = new ArrayList<>();
+            MergeTableNode node = new MergeTableNode();
+            Map<String, Object> attrs = new HashMap<>();
+            attrs.put("disabled",false);
+            node.setAttrs(attrs);
+            nodes.add(node);
+            try (MockedStatic<BeanUtil> beanUtilMockedStatic = mockStatic(BeanUtil.class)) {
+                ClientMongoOperator clientMongoOperator = mock(ClientMongoOperator.class);
+                when(clientMongoOperator.findOne(any(Query.class), anyString(), eq(TaskDto.class))).thenReturn(null);
+                beanUtilMockedStatic.when(() -> BeanUtil.getBean(ClientMongoOperator.class)).thenReturn(clientMongoOperator);
+                doCallRealMethod().when(hazelcastTaskService).cleanMergeNode(taskDto,"test",nodes);
+                hazelcastTaskService.cleanMergeNode(taskDto,"test",nodes);
+                verify(clientMongoOperator, times(0)).findOne(any(Query.class), anyString(), eq(TaskDto.class));
+            }
+        }
+
+        @Test
+        void test_ContainSyncProgresse(){
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+            taskDto.setType(TaskDto.TYPE_INITIAL_SYNC_CDC);
+            Map<String, Object> taskAttrs = new HashMap<>();
+            taskAttrs.put("syncProgress",new SyncProgress());
+            taskDto.setAttrs(taskAttrs);
+            List<Node> nodes = new ArrayList<>();
+            MergeTableNode node = new MergeTableNode();
+            Map<String, Object> attrs = new HashMap<>();
+            attrs.put("disabled",false);
+            node.setAttrs(attrs);
+            nodes.add(node);
+            try (MockedStatic<BeanUtil> beanUtilMockedStatic = mockStatic(BeanUtil.class)) {
+                ClientMongoOperator clientMongoOperator = mock(ClientMongoOperator.class);
+                when(clientMongoOperator.findOne(any(Query.class), anyString(), eq(TaskDto.class))).thenReturn(null);
+                beanUtilMockedStatic.when(() -> BeanUtil.getBean(ClientMongoOperator.class)).thenReturn(clientMongoOperator);
+                doCallRealMethod().when(hazelcastTaskService).cleanMergeNode(taskDto,"test",nodes);
+                hazelcastTaskService.cleanMergeNode(taskDto,"test",nodes);
+                verify(clientMongoOperator, times(0)).findOne(any(Query.class), anyString(), eq(TaskDto.class));
+            }
+        }
+
+    }
 }
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/constant/DateUtilTest.java b/iengine/iengine-common/src/test/java/com/tapdata/constant/DateUtilTest.java
new file mode 100644
index 0000000000..b8a045fea1
--- /dev/null
+++ b/iengine/iengine-common/src/test/java/com/tapdata/constant/DateUtilTest.java
@@ -0,0 +1,47 @@
+package com.tapdata.constant;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.TimeZone;
+
+/**
+ * 日期处理工具类测试
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/11 15:25 Create
+ */
+public class DateUtilTest {
+
+    @Nested
+    class GetTimeZoneTest {
+
+        @Test
+        void testNullArgs() {
+            TimeZone defZone = TimeZone.getDefault();
+            TimeZone timeZone = DateUtil.getTimeZone(null);
+            Assertions.assertEquals(defZone.getRawOffset(), timeZone.getRawOffset());
+        }
+
+        @Test
+        void testNormal() {
+            for (int i = -12; i <= 12; i++) {
+                TimeZone timeZone = DateUtil.getTimeZone(i);
+                Assertions.assertEquals(i * 3600000, timeZone.getRawOffset());
+            }
+        }
+
+        @Test
+        void testIllegal() {
+            int offset = -13;
+            TimeZone timeZone = DateUtil.getTimeZone(offset);
+            Assertions.assertEquals(offset * 3600000, timeZone.getRawOffset());
+
+            offset = 13;
+            timeZone = DateUtil.getTimeZone(offset);
+            Assertions.assertEquals(offset * 3600000, timeZone.getRawOffset());
+        }
+
+    }
+}
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/constant/MongodbUtilTest.java b/iengine/iengine-common/src/test/java/com/tapdata/constant/MongodbUtilTest.java
new file mode 100644
index 0000000000..b36b74cb7d
--- /dev/null
+++ b/iengine/iengine-common/src/test/java/com/tapdata/constant/MongodbUtilTest.java
@@ -0,0 +1,56 @@
+package com.tapdata.constant;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class MongodbUtilTest {
+    @DisplayName("test get mongoUri without additionalString")
+    @Test
+    void test1() {
+        Map<String, Object> config = new HashMap<>();
+        config.put("user", "testUser");
+        config.put("password", "testPassword");
+        config.put("host", "localhost");
+        config.put("database", "testDatabase");
+        String mongoUri = MongodbUtil.getUri(config);
+        assertEquals("mongodb://testUser:testPassword@localhost/testDatabase", mongoUri);
+    }
+
+    @DisplayName("test get mongoUri with additionalString")
+    @Test
+    void test2() {
+        Map<String, Object> config = new HashMap<>();
+        config.put("user", "testUser");
+        config.put("password", "testPassword");
+        config.put("host", "localhost");
+        config.put("database", "testDatabase");
+        config.put("additionalString", "authSource=admin");
+        String mongoUri = MongodbUtil.getUri(config);
+        assertEquals("mongodb://testUser:testPassword@localhost/testDatabase?authSource=admin", mongoUri);
+    }
+    @DisplayName("test get mongoUri with user and password")
+    @Test
+    void test3(){
+        Map<String, Object> config = new HashMap<>();
+        config.put("host", "localhost");
+        config.put("database", "testDatabase");
+        config.put("additionalString", "authSource=admin");
+        String mongoUri = MongodbUtil.getUri(config);
+        assertEquals("mongodb://localhost/testDatabase?authSource=admin", mongoUri);
+    }
+    @DisplayName("test get MongoUri By Uri")
+    @Test
+    void test4(){
+        Map<String, Object> config = new HashMap<>();
+        String uri="mongodb://localhost/testDatabase?authSource=admin";
+        config.put("isUri",true);
+        config.put("uri",uri);
+        String uriResult = MongodbUtil.getUri(config);
+        assertEquals(uri,uriResult);
+    }
+}
diff --git a/iengine/iengine-common/src/test/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtilTest.java b/iengine/iengine-common/src/test/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtilTest.java
index 46c65ae071..0e0f27addb 100644
--- a/iengine/iengine-common/src/test/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtilTest.java
+++ b/iengine/iengine-common/src/test/java/io/tapdata/flow/engine/V2/util/ExternalStorageUtilTest.java
@@ -6,7 +6,6 @@
 import com.tapdata.tm.commons.externalStorage.ExternalStorageDto;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.util.ReflectionTestUtils;
-
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
  class ExternalStorageUtilTest {
@@ -34,4 +33,5 @@ void getRocksDBConfigTest(){
         assertEquals(exceptData,actualData.getPath());
 
     }
+
 }
diff --git a/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java b/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
index f0ecebdba8..b2b42eead2 100644
--- a/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
+++ b/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
@@ -540,8 +540,6 @@ void beforeEach(){
             when(parms.getPass()).thenReturn("testPasswd");
             when(parms.getHost()).thenReturn("test@tapdata.io");
             when(parms.getPort()).thenReturn(465);
-            when(parms.getProxyHost()).thenReturn("smtp.test.cn");
-            when(parms.getProxyPort()).thenReturn(1025);
             when(parms.getFrom()).thenReturn("from@tapdata.io");
         }
         @Test
@@ -549,6 +547,8 @@ void beforeEach(){
         void test1(){
             try (MockedStatic<Transport> mb = Mockito
                     .mockStatic(Transport.class)) {
+                when(parms.getProxyHost()).thenReturn("smtp.test.cn");
+                when(parms.getProxyPort()).thenReturn(1025);
                 when(parms.getProtocol()).thenReturn("SSL");
                 mb.when(()->Transport.send(any(MimeMessage.class))).thenAnswer(invocationOnMock -> {return null;});
                 MailUtils.sendEmailForProxy(parms, adressees, title, content, flag);
@@ -563,6 +563,8 @@ void test2(){
             mockSlf4jLog(mailUtils, log);
             try (MockedStatic<Transport> mb = Mockito
                     .mockStatic(Transport.class)) {
+                when(parms.getProxyHost()).thenReturn("smtp.test.cn");
+                when(parms.getProxyPort()).thenReturn(1025);
                 RuntimeException e = new RuntimeException("test ex");
                 mb.when(()->Transport.send(any(MimeMessage.class))).thenThrow(e);
                 when(parms.getProtocol()).thenReturn("NO_PROTOCOL");
@@ -571,5 +573,31 @@ void test2(){
                 verify(log).error("mail send error：{}", "test ex", e);
             }
         }
+
+        @Test
+        @DisplayName("test sendEmailForProxy method no proxy")
+        void test3(){
+            try (MockedStatic<Transport> mb = Mockito
+                    .mockStatic(Transport.class)) {
+                when(parms.getProtocol()).thenReturn("SSL");
+                mb.when(()->Transport.send(any(MimeMessage.class))).thenAnswer(invocationOnMock -> {return null;});
+                MailUtils.sendEmailForProxy(parms, adressees, title, content, flag);
+                mb.verify(() -> Transport.send(any(MimeMessage.class)),new Times(1));
+            }
+        }
+
+        @Test
+        @DisplayName("test sendEmailForProxy method no proxy")
+        void test4(){
+            try (MockedStatic<Transport> mb = Mockito
+                    .mockStatic(Transport.class)) {
+                when(parms.getProxyHost()).thenReturn("smtp.test.cn");
+                when(parms.getProxyPort()).thenReturn(0);
+                when(parms.getProtocol()).thenReturn("SSL");
+                mb.when(()->Transport.send(any(MimeMessage.class))).thenAnswer(invocationOnMock -> {return null;});
+                MailUtils.sendEmailForProxy(parms, adressees, title, content, flag);
+                mb.verify(() -> Transport.send(any(MimeMessage.class)),new Times(1));
+            }
+        }
     }
 }
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
index dcc33175ec..b55638a534 100644
--- a/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
@@ -42,7 +42,7 @@ void test1(){
         List<String> tables=new ArrayList<>();
         tables.add("testTableId");
         cdcDelay.addHeartbeatTable(tables);
-        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer, TaskDto.SYNC_TYPE_SYNC);
+        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer);
         assertEquals(true,tapEvent instanceof HeartbeatEvent);
     }
     @DisplayName("test filterAndCalcDelay LOG_COLLECTOR Task")
@@ -59,8 +59,8 @@ void test2(){
         List<String> tables=new ArrayList<>();
         tables.add("testTableId");
         cdcDelay.addHeartbeatTable(tables);
-        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer, TaskDto.SYNC_TYPE_LOG_COLLECTOR);
-        assertEquals(true,tapEvent instanceof TapUpdateRecordEvent);
+        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer);
+        assertEquals(false,tapEvent instanceof TapUpdateRecordEvent);
     }
     @DisplayName("test filterAndCalcDelay not heartbeat event")
     @Test
@@ -75,7 +75,7 @@ void test3(){
         tapUpdateRecordEvent.setTableId("testTable");
         LongConsumer consumer=(time)->{
         };
-        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer, TaskDto.SYNC_TYPE_LOG_COLLECTOR);
+        TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer);
         assertEquals(tapUpdateRecordEvent,tapEvent);
     }
 
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/ConnHeartbeatUtilsTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/ConnHeartbeatUtilsTest.java
index 490501a3e5..889a3ae6f4 100644
--- a/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/ConnHeartbeatUtilsTest.java
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/ConnHeartbeatUtilsTest.java
@@ -3,13 +3,11 @@
 import com.tapdata.tm.commons.dag.AccessNodeTypeEnum;
 import com.tapdata.tm.commons.schema.DataSourceConnectionDto;
 import com.tapdata.tm.commons.schema.DataSourceDefinitionDto;
-import org.junit.jupiter.api.DisplayName;
-import org.junit.jupiter.api.Test;
+import io.tapdata.pdk.apis.entity.Capability;
+import org.junit.jupiter.api.*;
+import org.mockito.Mockito;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -21,54 +19,161 @@
 @DisplayName("Class ConnHeartbeatUtils Test")
 class ConnHeartbeatUtilsTest {
 
-	@Test
-	@DisplayName("Method generateConnections test")
-	void testGenerateConnections() {
-		DataSourceDefinitionDto dataSourceDefinitionDto = new DataSourceDefinitionDto();
-		String testId = "test_id";
-		String type = "test_type";
-		String pdkHash = "test_pdk_hash";
-		dataSourceDefinitionDto.setType(type);
-		dataSourceDefinitionDto.setPdkHash(pdkHash);
-
-		DataSourceConnectionDto dataSourceConnectionDto = ConnHeartbeatUtils.generateConnections(testId, dataSourceDefinitionDto);
-
-		assertNotNull(dataSourceConnectionDto);
-		assertEquals(ConnHeartbeatUtils.CONNECTION_NAME, dataSourceConnectionDto.getName());
-		assertEquals(DataSourceConnectionDto.STATUS_READY, dataSourceConnectionDto.getStatus());
-		assertEquals("source", dataSourceConnectionDto.getConnection_type());
-		assertEquals(CreateTypeEnum.System, dataSourceConnectionDto.getCreateType());
-		assertEquals(dataSourceDefinitionDto.getType(), dataSourceConnectionDto.getDatabase_type());
-		assertEquals(DataSourceDefinitionDto.PDK_TYPE, dataSourceConnectionDto.getPdkType());
-		assertEquals(dataSourceDefinitionDto.getPdkHash(), dataSourceConnectionDto.getPdkHash());
-		assertEquals(0, dataSourceConnectionDto.getRetry());
-		assertEquals(AccessNodeTypeEnum.AUTOMATIC_PLATFORM_ALLOCATION.name(), dataSourceConnectionDto.getAccessNodeType());
-		Map<String, Object> config = dataSourceConnectionDto.getConfig();
-		assertNotNull(config);
-		assertEquals(ConnHeartbeatUtils.MODE, config.get("mode"));
-		assertEquals(1, config.get("initial_totals"));
-		assertEquals(1000, config.get("incremental_interval"));
-		assertEquals(1, config.get("incremental_interval_totals"));
-		Object incrementalTypes = config.get("incremental_types");
-		assertInstanceOf(int[].class, incrementalTypes);
-		assertEquals(1, ((int[]) incrementalTypes).length);
-		assertEquals(1, ((int[]) incrementalTypes)[0]);
-		assertEquals(ConnHeartbeatUtils.TABLE_NAME, config.get("table_name"));
-		Object tableFieldsObj = config.get("table_fields");
-		assertInstanceOf(ArrayList.class, tableFieldsObj);
-		List<?> tableFields = (ArrayList<?>) tableFieldsObj;
-		assertEquals(2, tableFields.size());
-		Object obj1 = tableFields.get(0);
-		assertInstanceOf(HashMap.class, obj1);
-		Map<?, ?> map1 = (HashMap<?, ?>) obj1;
-		assertTrue((Boolean) map1.get("pri"));
-		assertEquals("id", map1.get("name"));
-		assertEquals("string(64)", map1.get("type"));
-		Object obj2 = tableFields.get(1);
-		assertInstanceOf(HashMap.class, obj2);
-		Map<?, ?> map2 = (HashMap<?, ?>) obj2;
-		assertFalse((Boolean) map2.get("pri"));
-		assertEquals("ts", map2.get("name"));
-		assertEquals("now", map2.get("type"));
-	}
-}
\ No newline at end of file
+    @Test
+    @DisplayName("Method generateConnections test")
+    void testGenerateConnections() {
+        DataSourceDefinitionDto dataSourceDefinitionDto = new DataSourceDefinitionDto();
+        String testId = "test_id";
+        String type = "test_type";
+        String pdkHash = "test_pdk_hash";
+        dataSourceDefinitionDto.setType(type);
+        dataSourceDefinitionDto.setPdkHash(pdkHash);
+
+        DataSourceConnectionDto dataSourceConnectionDto = ConnHeartbeatUtils.generateConnections(testId, dataSourceDefinitionDto);
+
+        assertNotNull(dataSourceConnectionDto);
+        assertEquals(ConnHeartbeatUtils.CONNECTION_NAME, dataSourceConnectionDto.getName());
+        assertEquals(DataSourceConnectionDto.STATUS_READY, dataSourceConnectionDto.getStatus());
+        assertEquals("source", dataSourceConnectionDto.getConnection_type());
+        assertEquals(CreateTypeEnum.System, dataSourceConnectionDto.getCreateType());
+        assertEquals(dataSourceDefinitionDto.getType(), dataSourceConnectionDto.getDatabase_type());
+        assertEquals(DataSourceDefinitionDto.PDK_TYPE, dataSourceConnectionDto.getPdkType());
+        assertEquals(dataSourceDefinitionDto.getPdkHash(), dataSourceConnectionDto.getPdkHash());
+        assertEquals(0, dataSourceConnectionDto.getRetry());
+        assertEquals(AccessNodeTypeEnum.AUTOMATIC_PLATFORM_ALLOCATION.name(), dataSourceConnectionDto.getAccessNodeType());
+        Map<String, Object> config = dataSourceConnectionDto.getConfig();
+        assertNotNull(config);
+        assertEquals(ConnHeartbeatUtils.MODE, config.get("mode"));
+        assertEquals(1, config.get("initial_totals"));
+        assertEquals(1000, config.get("incremental_interval"));
+        assertEquals(1, config.get("incremental_interval_totals"));
+        Object incrementalTypes = config.get("incremental_types");
+        assertInstanceOf(int[].class, incrementalTypes);
+        assertEquals(1, ((int[]) incrementalTypes).length);
+        assertEquals(1, ((int[]) incrementalTypes)[0]);
+        assertEquals(ConnHeartbeatUtils.TABLE_NAME, config.get("table_name"));
+        Object tableFieldsObj = config.get("table_fields");
+        assertInstanceOf(ArrayList.class, tableFieldsObj);
+        List<?> tableFields = (ArrayList<?>) tableFieldsObj;
+        assertEquals(2, tableFields.size());
+        Object obj1 = tableFields.get(0);
+        assertInstanceOf(HashMap.class, obj1);
+        Map<?, ?> map1 = (HashMap<?, ?>) obj1;
+        assertTrue((Boolean) map1.get("pri"));
+        assertEquals("id", map1.get("name"));
+        assertEquals("string(64)", map1.get("type"));
+        Object obj2 = tableFields.get(1);
+        assertInstanceOf(HashMap.class, obj2);
+        Map<?, ?> map2 = (HashMap<?, ?>) obj2;
+        assertFalse((Boolean) map2.get("pri"));
+        assertEquals("ts", map2.get("name"));
+        assertEquals("now", map2.get("type"));
+    }
+
+    @Nested
+    class HasNoHeartbeatTagTest {
+        @Test
+        void testNullTags() {
+            DataSourceConnectionDto connectionDto = Mockito.mock(DataSourceConnectionDto.class);
+            Mockito.when(connectionDto.getDefinitionTags()).thenReturn(null);
+            Assertions.assertFalse(ConnHeartbeatUtils.hasNoHeartbeatTag(connectionDto));
+        }
+
+        @Test
+        void testEmptyTags() {
+            DataSourceConnectionDto connectionDto = Mockito.mock(DataSourceConnectionDto.class);
+            Mockito.when(connectionDto.getDefinitionTags()).thenReturn(new ArrayList<>());
+            Assertions.assertFalse(ConnHeartbeatUtils.hasNoHeartbeatTag(connectionDto));
+        }
+
+        @Test
+        void testReturnTrue() {
+            DataSourceConnectionDto connectionDto = Mockito.mock(DataSourceConnectionDto.class);
+            Mockito.when(connectionDto.getDefinitionTags()).thenReturn(Collections.singletonList(ConnHeartbeatUtils.CONNECTOR_TAGS_NO_HEARTBEAT));
+            Assertions.assertTrue(ConnHeartbeatUtils.hasNoHeartbeatTag(connectionDto));
+        }
+    }
+
+    @Nested
+    class CheckConnectionTest {
+
+        DataSourceConnectionDto connectionDto;
+        List<Capability> capabilities;
+
+        @BeforeEach
+        void setUp() {
+            capabilities = new ArrayList<>();
+            capabilities.add(new Capability(CapabilityEnum.STREAM_READ_FUNCTION.name()));
+            capabilities.add(new Capability(CapabilityEnum.CREATE_TABLE_FUNCTION.name()));
+            capabilities.add(new Capability(CapabilityEnum.CREATE_TABLE_V2_FUNCTION.name()));
+            capabilities.add(new Capability(CapabilityEnum.WRITE_RECORD_FUNCTION.name()));
+            capabilities.add(new Capability(CapabilityEnum.DROP_TABLE_FUNCTION.name()));
+
+            connectionDto = new DataSourceConnectionDto();
+            connectionDto.setHeartbeatEnable(true);
+            connectionDto.setCapabilities(capabilities);
+            connectionDto.setDatabase_type(String.format("not-%s", ConnHeartbeatUtils.PDK_NAME));
+            connectionDto.setConnection_type("source_and_target");
+        }
+
+        @Test
+        void testDisableHeartbeats() {
+            connectionDto.setHeartbeatEnable(false);
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testDummyConnector() {
+            connectionDto.setDatabase_type(ConnHeartbeatUtils.PDK_NAME);
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testNullCapabilities() {
+            connectionDto.setCapabilities(null);
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testSourceConnection() {
+            connectionDto.setConnection_type("source");
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testTargetConnection() {
+            connectionDto.setConnection_type("target");
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testHasNoHeartbeatTag() {
+            connectionDto.setDefinitionTags(Collections.singletonList(ConnHeartbeatUtils.CONNECTOR_TAGS_NO_HEARTBEAT));
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testNoStreamReadFunction() {
+            capabilities.removeIf(capability -> capability.getId().equals(CapabilityEnum.STREAM_READ_FUNCTION.name()));
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testNoCreateTableFunction() {
+            capabilities.removeIf(capability -> capability.getId().equals(CapabilityEnum.CREATE_TABLE_FUNCTION.name()));
+            capabilities.removeIf(capability -> capability.getId().equals(CapabilityEnum.CREATE_TABLE_V2_FUNCTION.name()));
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testNoWriteRecordFunction() {
+            capabilities.removeIf(capability -> capability.getId().equals(CapabilityEnum.WRITE_RECORD_FUNCTION.name()));
+            Assertions.assertFalse(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+
+        @Test
+        void testTrue() {
+            Assertions.assertTrue(ConnHeartbeatUtils.checkConnection(connectionDto));
+        }
+    }
+}
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/metadatainstance/service/MetadataInstancesServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImplTest.java
index 738a65a1a3..043f43e26b 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/metadatainstance/service/MetadataInstancesServiceImplTest.java
@@ -2815,8 +2815,6 @@ void test1() {
 
 
 					Criteria criteria = Criteria.where(ConnHeartbeatUtils.TASK_RELATION_FIELD).is("65ae2d427f580c320ec3bc65");
-					criteria.and("status").ne("deleting");
-					criteria.and("is_deleted").ne(true);
 					Query queryHeartBeat = new Query(criteria);
 					queryHeartBeat.fields().include("_id", "dag");
 					heartbeatTaskDto.setDag(dag);
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;
+        }
+
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
index 76f6bacb7c..f43d214c74 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
@@ -5035,145 +5035,6 @@ void test2(){
         }
     }
 
-    @Nested
-    class checkHeartTableInLogCollectorTaskTableClass{
-        TaskDto taskDto;
-        UserDetail userDetail;
-        DAG dag;
-        DataSourceService dataSourceService;
-
-        ShareCdcTableMappingService shareCdcTableMappingService;
-        LogCollectorNode logCollectorNode;
-        List<Node> nodes=new ArrayList<>();
-        List<String> connectionIds=new ArrayList<>();
-        List<String> tableNames=new ArrayList<>();
-
-        @BeforeEach
-        void init() {
-            taskDto = mock(TaskDto.class);
-            dag = mock(DAG.class);
-            when(taskDto.getSyncType()).thenReturn(TaskDto.SYNC_TYPE_LOG_COLLECTOR);
-            when(taskDto.getDag()).thenReturn(dag);
-            taskDto.setSyncType(TaskDto.SYNC_TYPE_LOG_COLLECTOR);
-            userDetail = mock(UserDetail.class);
-            dataSourceService = mock(DataSourceServiceImpl.class);
-            shareCdcTableMappingService = mock(ShareCdcTableMappingService.class);
-            ReflectionTestUtils.setField(taskService, "shareCdcTableMappingService", shareCdcTableMappingService);
-            ReflectionTestUtils.setField(taskService, "dataSourceService", dataSourceService);
-
-            logCollectorNode = new LogCollectorNode();
-
-            nodes.add(logCollectorNode);
-
-
-            connectionIds.add("665eb04d9e541e4fee4f87c5");
-            logCollectorNode.setConnectionIds(connectionIds);
-
-
-            tableNames.add("testTableName");
-            logCollectorNode.setTableNames(tableNames);
-
-            Map<String, LogCollecotrConnConfig> logCollectorConnConfigs=new HashMap<>();
-            logCollectorNode.setLogCollectorConnConfigs(logCollectorConnConfigs);
-        }
-        @DisplayName("test update task config when tableConfig update")
-        @Test
-        void test1(){
-            when(dag.getSourceNodes()).thenReturn(nodes);
-            doCallRealMethod().when(taskService).checkHeartTableInLogCollectorTaskTable(taskDto, userDetail);
-            when(taskService.addHeartBeatTable2LogCollector(any(),any(),any())).thenReturn(false);
-            taskService.checkHeartTableInLogCollectorTaskTable(taskDto, userDetail);
-            verify(taskService,times(0)).updateById(taskDto,userDetail);
-        }
-        @DisplayName("test update task config when tableConfig not update")
-        @Test
-        void test2(){
-            LogCollectorNode logCollectorNode=new LogCollectorNode();
-            List<Node> nodes=new ArrayList<>();
-            nodes.add(logCollectorNode);
-
-            List<String> connectionIds=new ArrayList<>();
-            connectionIds.add("665eb04d9e541e4fee4f87c5");
-            logCollectorNode.setConnectionIds(connectionIds);
-
-            List<String> tableNames=new ArrayList<>();
-            tableNames.add("testTableName");
-            logCollectorNode.setTableNames(tableNames);
-
-            Map<String, LogCollecotrConnConfig> logCollectorConnConfigs=new HashMap<>();
-            logCollectorNode.setLogCollectorConnConfigs(logCollectorConnConfigs);
-
-            when(dag.getSourceNodes()).thenReturn(nodes);
-            doCallRealMethod().when(taskService).checkHeartTableInLogCollectorTaskTable(taskDto, userDetail);
-            when(taskService.addHeartBeatTable2LogCollector(any(),any(),any())).thenReturn(true);
-            taskService.checkHeartTableInLogCollectorTaskTable(taskDto, userDetail);
-            verify(taskService,times(1)).updateById(taskDto,userDetail);
-        }
-    }
-    @Nested
-    class testUpdateTableNameClass{
-        Set<String> tableNames;
-
-        @BeforeEach
-        void init() {
-            tableNames = new HashSet<>();
-            tableNames.add("testTable");
-        }
-        @DisplayName("test addHeartBeatTable2LogCollector method if connection enableHeartBeat")
-        @Test
-        void test1(){
-            List<DataSourceConnectionDto> dataSourceConnectionList=new ArrayList<>();
-            DataSourceConnectionDto dataSourceConnectionDto=new DataSourceConnectionDto();
-            dataSourceConnectionDto.setId(MongoUtils.toObjectId("665eb04d9e541e4fee4f87c5"));
-            dataSourceConnectionDto.setHeartbeatEnable(true);
-            dataSourceConnectionList.add(dataSourceConnectionDto);
-
-            Map<String, LogCollecotrConnConfig> logCollectorConnConfigs =new HashMap<>();
-            doCallRealMethod().when(taskService).addHeartBeatTable2LogCollector(dataSourceConnectionList,logCollectorConnConfigs,tableNames);
-            boolean updateConfig = taskService.addHeartBeatTable2LogCollector(dataSourceConnectionList, logCollectorConnConfigs, tableNames);
-            assertEquals(true,updateConfig);
-            assertEquals(2,tableNames.size());
-        }
-        @DisplayName("test addHeartBeatTable2LogCollector method if connection disableHeartBeat")
-        @Test
-        void test2(){
-            List<DataSourceConnectionDto> dataSourceConnectionList=new ArrayList<>();
-            DataSourceConnectionDto dataSourceConnectionDto=new DataSourceConnectionDto();
-            dataSourceConnectionDto.setId(MongoUtils.toObjectId("665eb04d9e541e4fee4f87c5"));
-            dataSourceConnectionDto.setHeartbeatEnable(false);
-            dataSourceConnectionList.add(dataSourceConnectionDto);
-
-            Map<String, LogCollecotrConnConfig> logCollectorConnConfigs =new HashMap<>();
-            doCallRealMethod().when(taskService).addHeartBeatTable2LogCollector(dataSourceConnectionList,logCollectorConnConfigs,tableNames);
-            boolean updateConfig = taskService.addHeartBeatTable2LogCollector(dataSourceConnectionList, logCollectorConnConfigs, tableNames);
-
-            assertEquals(false,updateConfig);
-            assertEquals(1,tableNames.size());
-        }
-        @Test
-        void test3(){
-            String connectionId="665eb04d9e541e4fee4f87c5";
-            List<DataSourceConnectionDto> dataSourceConnectionList=new ArrayList<>();
-            DataSourceConnectionDto dataSourceConnectionDto=new DataSourceConnectionDto();
-            dataSourceConnectionDto.setId(MongoUtils.toObjectId(connectionId));
-            dataSourceConnectionDto.setHeartbeatEnable(true);
-            dataSourceConnectionList.add(dataSourceConnectionDto);
-
-            Map<String, LogCollecotrConnConfig> logCollectorConnConfigs =new HashMap<>();
-
-            List<String> tableNamesList=new ArrayList<>();
-            tableNamesList.add("testTable");
-            LogCollecotrConnConfig logCollecotrConnConfig=new LogCollecotrConnConfig(connectionId, tableNamesList);
-            logCollectorConnConfigs.put(connectionId, logCollecotrConnConfig);
-
-            doCallRealMethod().when(taskService).addHeartBeatTable2LogCollector(dataSourceConnectionList,logCollectorConnConfigs,tableNames);
-            boolean updateConfig = taskService.addHeartBeatTable2LogCollector(dataSourceConnectionList, logCollectorConnConfigs, tableNames);
-            assertEquals(true,updateConfig);
-            assertEquals(2,tableNames.size());
-            assertEquals(2,logCollecotrConnConfig.getTableNames().size());
-        }
-
-    }
     @Nested
     class CheckSourceTimeDifferenceTest{
 
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
index 3aba8b48da..11ba910ad1 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
@@ -3,13 +3,20 @@
 
 import com.tapdata.tm.agent.service.AgentGroupService;
 import com.tapdata.tm.commons.dag.AccessNodeTypeEnum;
+import com.tapdata.tm.commons.dag.DAG;
+import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
+import com.tapdata.tm.commons.dag.nodes.TableNode;
+import com.tapdata.tm.commons.dag.vo.SyncObjects;
 import com.tapdata.tm.commons.schema.DataSourceConnectionDto;
 import com.tapdata.tm.commons.schema.MetadataInstancesDto;
 import com.tapdata.tm.commons.schema.Tag;
 import com.tapdata.tm.commons.schema.bean.SourceDto;
+import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.metadatainstance.service.MetadataInstancesService;
 import com.tapdata.tm.metadatainstance.service.MetadataInstancesServiceImpl;
+import com.tapdata.tm.task.service.TaskService;
 import com.tapdata.tm.worker.entity.Worker;
 import com.tapdata.tm.worker.service.WorkerService;
 import org.bson.types.ObjectId;
@@ -17,13 +24,12 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import java.lang.reflect.Array;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
+import java.util.*;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
 
@@ -85,7 +91,7 @@ String assertVerify(int getProcessIdTimes, int getProcessNodeListWithGroupTimes,
         @Test
         void testNormal() {
             String nodeId = assertVerify(1, 1, 1);
-            Assertions.assertEquals("id", nodeId);
+            assertEquals("id", nodeId);
         }
         @Test
         void testAvailableAgentIsNull() {
@@ -112,13 +118,13 @@ void testProcessIdsEmpty() {
         void testIsNotManually() {
             when(connectionDto.getAccessNodeType()).thenReturn(AccessNodeTypeEnum.AUTOMATIC_PLATFORM_ALLOCATION.name());
             String nodeId = assertVerify(1, 1, 1);
-            Assertions.assertEquals("id", nodeId);
+            assertEquals("id", nodeId);
         }
         @Test
         void testProcessNodeListWithGroupIsEmpty() {
             when(agentGroupService.getProcessNodeListWithGroup(connectionDto, user)).thenReturn(new ArrayList<>());
             String nodeId = assertVerify(1, 1, 1);
-            Assertions.assertEquals("id", nodeId);
+            assertEquals("id", nodeId);
         }
         @Test
         void testProcessNodeListWithGroupNotContainsId() {
@@ -185,4 +191,148 @@ void test_AncestorsNameIsNull(){
         }
 
     }
+    @Nested
+    class ldpTableStatusTest{
+        private TaskService taskService;
+
+        @BeforeEach
+        void setUp() {
+            taskService = mock(TaskService.class);
+            ReflectionTestUtils.setField(ldpService,"taskService",taskService);
+        }
+
+        @Test
+        void testLdpTableStatus_FDM_RunningState() {
+            String connectionId = "connection1";
+            List<String> tableNames = Arrays.asList("table1", "table2");
+            String ldpType = TaskDto.LDP_TYPE_FDM;
+            UserDetail user = mock(UserDetail.class);
+
+            TaskDto taskDto = mock(TaskDto.class);
+            DAG dag = mock(DAG.class);
+            when(taskDto.getDag()).thenReturn(dag);
+            when(taskDto.getStatus()).thenReturn(TaskDto.STATUS_RUNNING);
+            when(taskDto.getLdpNewTables()).thenReturn(Collections.emptyList());
+
+            DatabaseNode databaseNode = mock(DatabaseNode.class);
+            when(databaseNode.getConnectionId()).thenReturn(connectionId);
+            SyncObjects syncObjects = mock(SyncObjects.class);
+            LinkedHashMap<String, String> tableNameRelation = new LinkedHashMap<>();
+            tableNameRelation.put("FDM_table1","table1");
+            when(syncObjects.getTableNameRelation()).thenReturn(tableNameRelation);
+            when(databaseNode.getSyncObjects()).thenReturn(Collections.singletonList(syncObjects));
+
+            LinkedList<DatabaseNode> targetNode = new LinkedList<>();
+            targetNode.add(databaseNode);
+            when(dag.getTargetNode()).thenReturn(targetNode);
+
+            when(taskService.findAllDto(any(Query.class), eq(user)))
+                    .thenReturn(Collections.singletonList(taskDto));
+            when(metadataInstancesService.checkTableExist(anyString(), anyString(), eq(user)))
+                    .thenReturn(true);
+
+            doCallRealMethod().when(ldpService).ldpTableStatus(connectionId, tableNames, ldpType, user);
+            Map<String, String> result = ldpService.ldpTableStatus(connectionId, tableNames, ldpType, user);
+
+            assertEquals("running", result.get("table1"));
+            assertEquals("noRunning", result.get("table2"));
+        }
+
+        @Test
+        void testLdpTableStatus_FDM_NoRunningState() {
+            String connectionId = "connection2";
+            List<String> tableNames = Arrays.asList("table3");
+            String ldpType = TaskDto.LDP_TYPE_FDM;
+            UserDetail user = mock(UserDetail.class);
+
+            TaskDto taskDto = mock(TaskDto.class);
+            DAG dag = mock(DAG.class);
+            when(taskDto.getDag()).thenReturn(dag);
+            when(taskDto.getStatus()).thenReturn(TaskDto.STATUS_STOP);
+            when(taskDto.getLdpNewTables()).thenReturn(Collections.emptyList());
+
+            DatabaseNode databaseNode = mock(DatabaseNode.class);
+            when(databaseNode.getConnectionId()).thenReturn(connectionId);
+            SyncObjects syncObjects = mock(SyncObjects.class);
+            LinkedHashMap<String, String> tableNameRelation = new LinkedHashMap<>();
+            tableNameRelation.put("FDM_table3","table3");
+            when(syncObjects.getTableNameRelation()).thenReturn(tableNameRelation);
+            when(databaseNode.getSyncObjects()).thenReturn(Collections.singletonList(syncObjects));
+
+
+            LinkedList<DatabaseNode> targetNode = new LinkedList<>();
+            targetNode.add(databaseNode);
+            when(dag.getTargetNode()).thenReturn(targetNode);
+
+            when(taskService.findAllDto(any(Query.class), eq(user)))
+                    .thenReturn(Collections.singletonList(taskDto));
+            when(metadataInstancesService.checkTableExist(anyString(), anyString(), eq(user)))
+                    .thenReturn(false);
+            doCallRealMethod().when(ldpService).ldpTableStatus(connectionId, tableNames, ldpType, user);
+            Map<String, String> result = ldpService.ldpTableStatus(connectionId, tableNames, ldpType, user);
+
+            assertEquals("noRunning", result.get("table3"));
+        }
+        @Test
+        void testLdpTableStatus_FDM_CompleteState() {
+            String connectionId = "connection2";
+            List<String> tableNames = Arrays.asList("table3");
+            String ldpType = TaskDto.LDP_TYPE_FDM;
+            UserDetail user = mock(UserDetail.class);
+
+            TaskDto taskDto = mock(TaskDto.class);
+            DAG dag = mock(DAG.class);
+            when(taskDto.getDag()).thenReturn(dag);
+            when(taskDto.getStatus()).thenReturn(TaskDto.STATUS_COMPLETE);
+            when(taskDto.getLdpNewTables()).thenReturn(Collections.emptyList());
+
+            DatabaseNode databaseNode = mock(DatabaseNode.class);
+            when(databaseNode.getConnectionId()).thenReturn(connectionId);
+            SyncObjects syncObjects = mock(SyncObjects.class);
+            LinkedHashMap<String, String> tableNameRelation = new LinkedHashMap<>();
+            tableNameRelation.put("FDM_table3","table3");
+            when(syncObjects.getTableNameRelation()).thenReturn(tableNameRelation);
+            when(databaseNode.getSyncObjects()).thenReturn(Collections.singletonList(syncObjects));
+
+
+            LinkedList<DatabaseNode> targetNode = new LinkedList<>();
+            targetNode.add(databaseNode);
+            when(dag.getTargetNode()).thenReturn(targetNode);
+
+            when(taskService.findAllDto(any(Query.class), eq(user)))
+                    .thenReturn(Collections.singletonList(taskDto));
+            doCallRealMethod().when(ldpService).ldpTableStatus(connectionId, tableNames, ldpType, user);
+            Map<String, String> result = ldpService.ldpTableStatus(connectionId, tableNames, ldpType, user);
+
+            assertEquals("running", result.get("table3"));
+        }
+        @Test
+        void testLdpTableStatus_MDM_CompleteState() {
+            String connectionId = "connection2";
+            List<String> tableNames = Arrays.asList("table3");
+            String ldpType = TaskDto.LDP_TYPE_MDM;
+            UserDetail user = mock(UserDetail.class);
+
+            TaskDto taskDto = mock(TaskDto.class);
+            DAG dag = mock(DAG.class);
+            when(taskDto.getDag()).thenReturn(dag);
+            when(taskDto.getStatus()).thenReturn(TaskDto.STATUS_COMPLETE);
+            when(taskDto.getLdpNewTables()).thenReturn(Collections.emptyList());
+
+            TableNode tableNode = mock(TableNode.class);
+            when(tableNode.getConnectionId()).thenReturn(connectionId);
+            when(tableNode.getTableName()).thenReturn("table3");
+
+            List<Node> targetNode = new LinkedList<>();
+            targetNode.add(tableNode);
+            when(dag.getTargets()).thenReturn(targetNode);
+
+            when(taskService.findAllDto(any(Query.class), eq(user)))
+                    .thenReturn(Collections.singletonList(taskDto));
+            doCallRealMethod().when(ldpService).ldpTableStatus(connectionId, tableNames, ldpType, user);
+            Map<String, String> result = ldpService.ldpTableStatus(connectionId, tableNames, ldpType, user);
+
+            assertEquals("running", result.get("table3"));
+        }
+    }
 }
\ No newline at end of file
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskDagServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskDagServiceImplTest.java
new file mode 100644
index 0000000000..615bd219cd
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskDagServiceImplTest.java
@@ -0,0 +1,97 @@
+package com.tapdata.tm.task.service.impl;
+
+import com.tapdata.tm.commons.dag.DAG;
+import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
+import com.tapdata.tm.commons.dag.process.MigrateFieldRenameProcessorNode;
+import com.tapdata.tm.commons.dag.process.MigrateJsProcessorNode;
+import com.tapdata.tm.commons.dag.process.TableRenameProcessNode;
+import com.tapdata.tm.commons.dag.vo.SyncObjects;
+import com.tapdata.tm.commons.dag.vo.TableFieldInfo;
+import com.tapdata.tm.commons.dag.vo.TableRenameTableInfo;
+import com.tapdata.tm.commons.task.dto.TaskDto;
+import org.elasticsearch.common.collect.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+
+import java.util.LinkedHashSet;
+import java.util.LinkedList;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+public class TaskDagServiceImplTest {
+    @Nested
+    class calculationDagHashTest {
+        private TaskDto taskDto;
+
+        private DAG dag;
+
+        private DatabaseNode sourceNode;
+
+        private TableRenameProcessNode tableRenameNode;
+
+        private MigrateFieldRenameProcessorNode fieldRenameNode;
+
+        private MigrateJsProcessorNode jsProcessorNode;
+
+        private TaskDagServiceImpl taskDagServiceImpl;
+
+        private LinkedList<DatabaseNode> sourceNodes;
+        private LinkedList<Node> dagNodes;
+
+        @BeforeEach
+        public void setUp() {
+            taskDto = mock(TaskDto.class);
+            dag = mock(DAG.class);
+            taskDagServiceImpl = spy(TaskDagServiceImpl.class);
+            sourceNode = mock(DatabaseNode.class);
+            tableRenameNode = mock(TableRenameProcessNode.class);
+            fieldRenameNode = mock(MigrateFieldRenameProcessorNode.class);
+            jsProcessorNode = mock(MigrateJsProcessorNode.class);
+            sourceNodes = new LinkedList<>();
+            dagNodes = new LinkedList<>();
+            sourceNodes.add(sourceNode);
+
+            when(taskDto.getDag()).thenReturn(dag);
+            when(dag.getSourceNode()).thenReturn(sourceNodes);
+            when(sourceNode.getTableNames()).thenReturn(List.of("table1", "table2"));
+            when(dag.getNodes()).thenReturn(dagNodes);
+        }
+
+        @Test
+        public void testCalculationDagHash_Success() {
+            LinkedHashSet<TableRenameTableInfo> renameInfos = new LinkedHashSet<>();
+            renameInfos.add(mock(TableRenameTableInfo.class));
+            when(tableRenameNode.getTableNames()).thenReturn(renameInfos);
+            dagNodes.add(tableRenameNode);
+
+            LinkedList<TableFieldInfo> fieldInfos = new LinkedList<>();
+            fieldInfos.add(mock(TableFieldInfo.class));
+            when(fieldRenameNode.getFieldsMapping()).thenReturn(fieldInfos);
+            dagNodes.add(fieldRenameNode);
+
+            when(jsProcessorNode.getScript()).thenReturn("script content");
+            when(jsProcessorNode.getDeclareScript()).thenReturn("declare content");
+            dagNodes.add(jsProcessorNode);
+
+            LinkedList<DatabaseNode> targetNodes = new LinkedList<>();
+            DatabaseNode targetNode = mock(DatabaseNode.class);
+            when(dag.getTargetNode()).thenReturn(targetNodes);
+            when(targetNode.getSyncObjects()).thenReturn(List.of(mock(SyncObjects.class)));
+            targetNodes.add(targetNode);
+
+            int hash = taskDagServiceImpl.calculationDagHash(taskDto);
+            assertNotEquals(0, hash, "Hash should not be zero");
+        }
+
+        @Test
+        public void testCalculationDagHash_RandomReturn() {
+            int randomHash = taskDagServiceImpl.calculationDagHash(taskDto);
+            assertNotEquals(0, randomHash);
+        }
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/worker/service/WorkerServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/worker/service/WorkerServiceTest.java
index 29277bdd0c..8c922e64a9 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/worker/service/WorkerServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/worker/service/WorkerServiceTest.java
@@ -6,6 +6,7 @@
 import com.tapdata.tm.Settings.constant.SettingUtil;
 import com.tapdata.tm.Settings.entity.Settings;
 import com.tapdata.tm.Settings.service.SettingsService;
+import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.base.reporitory.BaseRepository;
 import com.tapdata.tm.cluster.service.ClusterStateService;
 import com.tapdata.tm.commons.base.dto.SchedulableDto;
@@ -16,16 +17,20 @@
 import com.tapdata.tm.permissions.IDataPermissionHelper;
 import com.tapdata.tm.permissions.service.DataPermissionService;
 import com.tapdata.tm.task.service.TaskService;
+import com.tapdata.tm.user.service.UserService;
 import com.tapdata.tm.utils.MongoUtils;
 import com.tapdata.tm.worker.dto.WorkerDto;
+import com.tapdata.tm.worker.dto.WorkerExpireDto;
 import com.tapdata.tm.worker.dto.WorkerProcessInfoDto;
 import com.tapdata.tm.worker.entity.Worker;
+import com.tapdata.tm.worker.entity.WorkerExpire;
 import com.tapdata.tm.worker.repository.WorkerRepository;
 import com.tapdata.tm.worker.vo.CalculationEngineVo;
 import org.bson.BsonValue;
 import org.junit.jupiter.api.*;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
+import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
@@ -379,4 +384,75 @@ void testEngineVersionDASS(){
             Assertions.assertTrue(workerService.checkEngineVersion(mock(UserDetail.class)));
         }
     }
+    @Nested
+    class createShareWorkerTest {
+        private WorkerExpireDto workerExpireDto;
+        private UserDetail loginUser;
+        private UserService userService;
+        private MongoTemplate mongoTemplate;
+        @BeforeEach
+        public void setUp() {
+            workerExpireDto = new WorkerExpireDto();
+            workerExpireDto.setUserId("testUser");
+            workerExpireDto.setSubscribeId("subscribe123");
+
+            loginUser = mock(UserDetail.class);
+            loginUser.setUserId("loginUser");
+            loginUser.setUsername("loginUsername");
+            mongoTemplate = mock(MongoTemplate.class);
+            userService = mock(UserService.class);
+            ReflectionTestUtils.setField(workerService, "userService", userService);
+            ReflectionTestUtils.setField(workerService, "mongoTemplate", mongoTemplate);
+            doCallRealMethod().when(workerService).createShareWorker(workerExpireDto, loginUser);
+        }
+
+        @Test
+        public void testCreateShareWorker_Success() {
+            when(settingsService.getValueByCategoryAndKey(CategoryEnum.SYSTEM, KeyEnum.SHARE_AGENT_EXPRIRE_DAYS))
+                    .thenReturn("30");
+            when(settingsService.getValueByCategoryAndKey(CategoryEnum.SYSTEM, KeyEnum.SHARE_AGENT_CREATE_USER))
+                    .thenReturn("agentUser1,agentUser2");
+
+            UserDetail sharedUserDetail = mock(UserDetail.class);
+            sharedUserDetail.setUserId("sharedUserId");
+            sharedUserDetail.setTcmUserId("sharedTcmUserId");
+            when(userService.loadUserByUsername(anyString()))
+                    .thenReturn(sharedUserDetail);
+            when(mongoTemplate.findOne(any(), eq(WorkerExpire.class)))
+                    .thenReturn(null);
+            workerService.createShareWorker(workerExpireDto, loginUser);
+            verify(mongoTemplate, times(1)).insert(any(WorkerExpire.class));
+            verify(userService, times(1)).loadUserByUsername(anyString());
+        }
+
+        @Test
+        public void testCreateShareWorker_UserAlreadyExists() {
+            WorkerExpire existingWorkerExpire = new WorkerExpire();
+            when(mongoTemplate.findOne(any(), eq(WorkerExpire.class)))
+                    .thenReturn(existingWorkerExpire);
+            BizException exception = assertThrows(BizException.class, () -> {
+                workerService.createShareWorker(workerExpireDto, loginUser);
+            });
+
+            assertEquals("SHARE_AGENT_USER_EXISTED", exception.getErrorCode());
+            assertEquals("have applied for a public agent", exception.getMessage());
+            verify(mongoTemplate, never()).insert(any(WorkerExpire.class));
+        }
+
+        @Test
+        public void testCreateShareWorker_UserNotFound() {
+            when(settingsService.getValueByCategoryAndKey(CategoryEnum.SYSTEM, KeyEnum.SHARE_AGENT_EXPRIRE_DAYS))
+                    .thenReturn("30");
+            when(settingsService.getValueByCategoryAndKey(CategoryEnum.SYSTEM, KeyEnum.SHARE_AGENT_CREATE_USER))
+                    .thenReturn("agentUser1,agentUser2");
+            when(userService.loadUserByUsername(anyString()))
+                    .thenReturn(null);
+            IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> {
+                workerService.createShareWorker(workerExpireDto, loginUser);
+            });
+
+            assertEquals("userDetail is null", exception.getMessage());
+            verify(mongoTemplate, never()).insert(any(WorkerExpire.class));
+        }
+    }
 }