Repository: tapdata/tapdata (language: python)

Issue:
Test: TAP-3437 Add constants 'user_id'.



Discussion / hints:

Fix patch:
diff --git a/RELEASE b/RELEASE
index ff0e70c2bf..e72fe6ea4c 100644
--- a/RELEASE
+++ b/RELEASE
@@ -1,2 +1,2 @@
-LST: v3.7.0
-DEV: v3.8.0-dev
\ No newline at end of file
+LST: v3.8.0
+DEV: v3.9.0-dev
\ No newline at end of file
diff --git a/iengine/api/pom.xml b/iengine/api/pom.xml
index 426cf0c2ec..2162f138b8 100644
--- a/iengine/api/pom.xml
+++ b/iengine/api/pom.xml
@@ -78,6 +78,11 @@
 				<version>4.11.0</version>
 				<scope>test</scope>
 			</dependency>
+            <dependency>
+                <groupId>org.springframework</groupId>
+                <artifactId>spring-test</artifactId>
+                <scope>test</scope>
+            </dependency>
     </dependencies>
 
     <distributionManagement>
diff --git a/iengine/api/src/main/java/io/tapdata/aspect/CreateTableFuncAspect.java b/iengine/api/src/main/java/io/tapdata/aspect/CreateTableFuncAspect.java
index b75e35044b..495f650678 100644
--- a/iengine/api/src/main/java/io/tapdata/aspect/CreateTableFuncAspect.java
+++ b/iengine/api/src/main/java/io/tapdata/aspect/CreateTableFuncAspect.java
@@ -6,6 +6,7 @@
 
 public class CreateTableFuncAspect extends DataFunctionAspect<CreateTableFuncAspect> {
 	private CreateTableOptions createTableOptions;
+	private boolean init;
 	public CreateTableFuncAspect createTableOptions(CreateTableOptions createTableOptions) {
 		this.createTableOptions = createTableOptions;
 		return this;
@@ -45,4 +46,13 @@ public CreateTableOptions getCreateTableOptions() {
 	public void setCreateTableOptions(CreateTableOptions createTableOptions) {
 		this.createTableOptions = createTableOptions;
 	}
+
+	public boolean isInit() {
+		return init;
+	}
+
+	public CreateTableFuncAspect setInit(boolean init) {
+		this.init = init;
+		return this;
+	}
 }
diff --git a/iengine/api/src/main/java/io/tapdata/aspect/DropTableFuncAspect.java b/iengine/api/src/main/java/io/tapdata/aspect/DropTableFuncAspect.java
index 276674622f..a960f293ad 100644
--- a/iengine/api/src/main/java/io/tapdata/aspect/DropTableFuncAspect.java
+++ b/iengine/api/src/main/java/io/tapdata/aspect/DropTableFuncAspect.java
@@ -5,6 +5,7 @@
 
 public class DropTableFuncAspect extends DataFunctionAspect<DropTableFuncAspect> {
 	private TapDropTableEvent dropTableEvent;
+	private boolean init;
 	public DropTableFuncAspect dropTableEvent(TapDropTableEvent dropTableEvent) {
 		this.dropTableEvent = dropTableEvent;
 		return this;
@@ -31,4 +32,13 @@ public TapDropTableEvent getDropTableEvent() {
 	public void setDropTableEvent(TapDropTableEvent dropTableEvent) {
 		this.dropTableEvent = dropTableEvent;
 	}
+
+	public boolean isInit() {
+		return init;
+	}
+
+	public DropTableFuncAspect setInit(boolean init) {
+		this.init = init;
+		return this;
+	}
 }
diff --git a/iengine/api/src/main/java/io/tapdata/aspect/task/impl/TaskSessionClassHolder.java b/iengine/api/src/main/java/io/tapdata/aspect/task/impl/TaskSessionClassHolder.java
index a337ce364b..3bccb0d43b 100644
--- a/iengine/api/src/main/java/io/tapdata/aspect/task/impl/TaskSessionClassHolder.java
+++ b/iengine/api/src/main/java/io/tapdata/aspect/task/impl/TaskSessionClassHolder.java
@@ -206,13 +206,15 @@ public void ensureTaskSessionStopped(TaskStopAspect stopAspect) {
 	public synchronized void ensureTaskSessionStopped(String taskId, TaskStopAspect stopAspect) {
 		if(taskId == null)
 			return;
-		AspectTaskEx aspectTask = aspectTaskMap.remove(taskId);
+		AspectTaskEx aspectTask = aspectTaskMap.get(taskId);
 		if (aspectTask != null) {
 			if(ignoreErrors)
 				CommonUtils.ignoreAnyError(() -> aspectTask.aspectTask.onStop(stopAspect), TAG);
 			else
 				CommonUtils.handleAnyError(() -> aspectTask.aspectTask.onStop(stopAspect));
 
+		    aspectTaskMap.remove(taskId);
+
 			List<Class<? extends Aspect>> observerClasses = aspectTask.aspectTask.observeAspects();
 			AspectManager aspectManager = InstanceFactory.instance(AspectManager.class);
 			if (observerClasses != null && !observerClasses.isEmpty()) {
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/AutoRecovery.java b/iengine/api/src/main/java/io/tapdata/inspect/AutoRecovery.java
new file mode 100644
index 0000000000..bb48bf412d
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/AutoRecovery.java
@@ -0,0 +1,83 @@
+package io.tapdata.inspect;
+
+import com.tapdata.entity.TapdataRecoveryEvent;
+import io.tapdata.inspect.exception.DuplicateAutoRecoveryException;
+import io.tapdata.inspect.exception.DuplicateClientAutoRecoveryException;
+import io.tapdata.inspect.exception.NotfoundAutoRecoveryException;
+import lombok.Getter;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+/**
+ * 数据修复操作入口
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/5 18:50 Create
+ */
+public class AutoRecovery implements AutoCloseable {
+    private static final Logger logger = LogManager.getLogger(AutoRecovery.class);
+    private static final Map<String, AutoRecovery> instances = new HashMap<>();
+
+    @Getter
+    private final String taskId;
+    private final Map<String, AutoRecoveryClient> clients = new ConcurrentHashMap<>();
+    private Consumer<TapdataRecoveryEvent> enqueueConsumer;
+
+    protected AutoRecovery(String taskId) {
+        this.taskId = taskId;
+    }
+
+    @Override
+    @SuppressWarnings("resource")
+    public void close() throws Exception {
+        logger.info("Releasing auto-recovery instance '{}'", getTaskId());
+        instances.remove(getTaskId());
+    }
+
+    public static AutoRecovery init(String taskId) {
+        return instances.compute(taskId, (id, ins) -> {
+            DuplicateAutoRecoveryException.assertNull(ins);
+            logger.info("Init auto-recovery instance '{}'", taskId);
+            return new AutoRecovery(taskId);
+        });
+    }
+
+    @SuppressWarnings("resource")
+    public static AutoRecoveryClient initClient(String taskId, String inspectTaskId, Consumer<TapdataRecoveryEvent> completedConsumer) {
+        AutoRecovery autoRecovery = instances.computeIfAbsent(taskId, NotfoundAutoRecoveryException::failed);
+        return autoRecovery.clients.compute(inspectTaskId, (id, client) -> {
+            DuplicateClientAutoRecoveryException.assertNull(client);
+            return new AutoRecoveryClient(taskId, id, autoRecovery.enqueueConsumer, completedConsumer) {
+                @Override
+                public void close() throws Exception {
+                    autoRecovery.clients.remove(id);
+                }
+            };
+        });
+    }
+
+    @SuppressWarnings("resource")
+    public static void setEnqueueConsumer(String taskId, Consumer<TapdataRecoveryEvent> enqueueConsumer) {
+        instances.computeIfPresent(taskId, (id, ins) -> {
+            logger.info("Set auto-recovery enqueue: '{}-{}'", taskId, id);
+            ins.enqueueConsumer = enqueueConsumer;
+            return ins;
+        });
+    }
+
+    @SuppressWarnings("resource")
+    public static void completed(String taskId, TapdataRecoveryEvent event) {
+        instances.computeIfPresent(taskId, (id, ins) -> {
+            ins.clients.computeIfPresent(event.getInspectTaskId(), (inspectTaskId, autoRecoveryClient) -> {
+                autoRecoveryClient.completed(event);
+                return autoRecoveryClient;
+            });
+            return ins;
+        });
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/AutoRecoveryClient.java b/iengine/api/src/main/java/io/tapdata/inspect/AutoRecoveryClient.java
new file mode 100644
index 0000000000..71d8c8bfb5
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/AutoRecoveryClient.java
@@ -0,0 +1,40 @@
+package io.tapdata.inspect;
+
+import com.tapdata.entity.TapdataRecoveryEvent;
+import lombok.Getter;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.function.Consumer;
+
+/**
+ * 自动修复数据客户端
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/13 19:40 Create
+ */
+public abstract class AutoRecoveryClient implements AutoCloseable {
+    private static final Logger logger = LogManager.getLogger(AutoRecoveryClient.class);
+    @Getter
+    private final String taskId;
+    @Getter
+    private final String inspectTaskId;
+    private final Consumer<TapdataRecoveryEvent> enqueueConsumer;
+    private final Consumer<TapdataRecoveryEvent> completedConsumer;
+
+    protected AutoRecoveryClient(String taskId, String inspectTaskId, Consumer<TapdataRecoveryEvent> enqueueConsumer, Consumer<TapdataRecoveryEvent> completedConsumer) {
+        this.taskId = taskId;
+        this.inspectTaskId = inspectTaskId;
+        this.enqueueConsumer = enqueueConsumer;
+        this.completedConsumer = completedConsumer;
+        logger.info("Init auto-recovery client: '{}-{}'", getTaskId(), getInspectTaskId());
+    }
+
+    public void enqueue(TapdataRecoveryEvent event) {
+        enqueueConsumer.accept(event);
+    }
+
+    public void completed(TapdataRecoveryEvent event) {
+        completedConsumer.accept(event);
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/exception/AutoRecoveryException.java b/iengine/api/src/main/java/io/tapdata/inspect/exception/AutoRecoveryException.java
new file mode 100644
index 0000000000..9c78f8186a
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/exception/AutoRecoveryException.java
@@ -0,0 +1,24 @@
+package io.tapdata.inspect.exception;
+
+/**
+ * 自动修复数据异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/21 11:40 Create
+ */
+public class AutoRecoveryException extends RuntimeException {
+    public AutoRecoveryException() {
+    }
+
+    public AutoRecoveryException(String message) {
+        super(message);
+    }
+
+    public AutoRecoveryException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public AutoRecoveryException(Throwable cause) {
+        super(cause);
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateAutoRecoveryException.java b/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateAutoRecoveryException.java
new file mode 100644
index 0000000000..28994243ab
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateAutoRecoveryException.java
@@ -0,0 +1,27 @@
+package io.tapdata.inspect.exception;
+
+import io.tapdata.inspect.AutoRecovery;
+import lombok.Getter;
+
+/**
+ * 自动修复数据实例重复异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/21 11:42 Create
+ */
+@Getter
+public class DuplicateAutoRecoveryException extends AutoRecoveryException {
+
+    private final String taskId;
+
+    public DuplicateAutoRecoveryException(String taskId) {
+        super(String.format("Task %s auto-recovery instance already exists.", taskId));
+        this.taskId = taskId;
+    }
+
+    public static void assertNull(AutoRecovery ins) {
+        if (null != ins) {
+            throw new DuplicateAutoRecoveryException(ins.getTaskId());
+        }
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateClientAutoRecoveryException.java b/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateClientAutoRecoveryException.java
new file mode 100644
index 0000000000..3fd75dc406
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/exception/DuplicateClientAutoRecoveryException.java
@@ -0,0 +1,29 @@
+package io.tapdata.inspect.exception;
+
+import io.tapdata.inspect.AutoRecoveryClient;
+import lombok.Getter;
+
+/**
+ * 自动修复客户端异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/21 11:42 Create
+ */
+@Getter
+public class DuplicateClientAutoRecoveryException extends AutoRecoveryException {
+
+    private final String taskId;
+    private final String inspectTaskId;
+
+    public DuplicateClientAutoRecoveryException(String taskId, String inspectTaskId) {
+        super(String.format("Task '%s-%s' auto-recovery client already exists.", taskId, inspectTaskId));
+        this.taskId = taskId;
+        this.inspectTaskId = inspectTaskId;
+    }
+
+    public static void assertNull(AutoRecoveryClient ins) {
+        if (null != ins) {
+            throw new DuplicateClientAutoRecoveryException(ins.getTaskId(), ins.getInspectTaskId());
+        }
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/exception/NotfoundAutoRecoveryException.java b/iengine/api/src/main/java/io/tapdata/inspect/exception/NotfoundAutoRecoveryException.java
new file mode 100644
index 0000000000..4d35b3b217
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/exception/NotfoundAutoRecoveryException.java
@@ -0,0 +1,24 @@
+package io.tapdata.inspect.exception;
+
+import lombok.Getter;
+
+/**
+ * 自动修复实例不存在异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/21 12:22 Create
+ */
+@Getter
+public class NotfoundAutoRecoveryException extends AutoRecoveryException {
+
+    private final String taskId;
+
+    public NotfoundAutoRecoveryException(String taskId) {
+        super(String.format("Not found auto-recovery instance by taskId: %s", taskId));
+        this.taskId = taskId;
+    }
+
+    public static <T> T failed(String taskId) {
+        throw new NotfoundAutoRecoveryException(taskId);
+    }
+}
diff --git a/iengine/api/src/main/java/io/tapdata/inspect/exception/TimeoutAutoRecoveryException.java b/iengine/api/src/main/java/io/tapdata/inspect/exception/TimeoutAutoRecoveryException.java
new file mode 100644
index 0000000000..9f74a21186
--- /dev/null
+++ b/iengine/api/src/main/java/io/tapdata/inspect/exception/TimeoutAutoRecoveryException.java
@@ -0,0 +1,26 @@
+package io.tapdata.inspect.exception;
+
+import lombok.Getter;
+
+/**
+ * 自动修复超时异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/21 12:22 Create
+ */
+@Getter
+public class TimeoutAutoRecoveryException extends AutoRecoveryException {
+
+    private final String taskId;
+
+    public TimeoutAutoRecoveryException(String taskId) {
+        super(String.format("Timeout auto-recovery task '%s'", taskId));
+        this.taskId = taskId;
+    }
+
+    public static void assertFalse(String taskId, boolean expected) {
+        if (expected) {
+            throw new TimeoutAutoRecoveryException(taskId);
+        }
+    }
+}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/common/DAGDataEngineServiceImpl.java b/iengine/iengine-app/src/main/java/io/tapdata/common/DAGDataEngineServiceImpl.java
index 118cdc3d25..185e9eacfe 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/common/DAGDataEngineServiceImpl.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/common/DAGDataEngineServiceImpl.java
@@ -31,6 +31,7 @@
 import io.tapdata.flow.engine.V2.task.TaskService;
 import io.tapdata.observable.logging.ObsLogger;
 import io.tapdata.observable.logging.ObsLoggerFactory;
+import io.tapdata.pdk.core.utils.CommonUtils;
 import io.tapdata.schema.TapTableMap;
 import io.tapdata.websocket.handler.DeduceSchemaHandler;
 import org.apache.commons.collections.CollectionUtils;
@@ -44,6 +45,7 @@
 
 
 public class DAGDataEngineServiceImpl extends DAGDataServiceImpl {
+    public static final String TAG = DAGDataEngineServiceImpl.class.getSimpleName();
     private final ObsLogger obsLogger;
     private final TaskService<TaskDto> taskService;
     private final TaskDto taskDto;
@@ -82,12 +84,12 @@ public DAGDataEngineServiceImpl(DeduceSchemaHandler.DeduceSchemaRequest request,
 
     @Override
     public TapTable loadTapTable(String nodeId, String virtualId, TaskDto taskDto) {
-        
+        TaskClient<TaskDto> taskClient = null;
         try {
             // 跑任务加载js模型
             String schemaKey = taskDto.getId() + "-" + virtualId;
             long startTs = System.currentTimeMillis();
-            TaskClient<TaskDto> taskClient = execTask(taskDto);
+            taskClient = execTask(taskDto);
 
             obsLogger.info("load tapTable task {} {}, cost {}ms", schemaKey, taskClient.getStatus(), (System.currentTimeMillis() - startTs));
             //成功
@@ -98,17 +100,21 @@ public TapTable loadTapTable(String nodeId, String virtualId, TaskDto taskDto) {
             return tapTable;
         } catch (Exception e) {
             obsLogger.error("An error occurred while obtaining the results of model deduction", e);
+        } finally {
+            TaskClient<TaskDto> finalTaskClient = taskClient;
+            CommonUtils.ignoreAnyError(() -> Optional.ofNullable(finalTaskClient).ifPresent(TaskClient::close), TAG + "-closeTaskClient");
         }
         return null;
     }
 
     @Override
     public List<MigrateJsResultVo> getJsResult(String jsNodeId, String virtualTargetId, TaskDto taskDto) {
+        TaskClient<TaskDto> taskClient = null;
         try {
             String schemaKey = taskDto.getId() + "-" + virtualTargetId;
             long startTs = System.currentTimeMillis();
 
-            TaskClient<TaskDto> taskClient = execTask(taskDto);
+            taskClient = execTask(taskDto);
 
             obsLogger.info("load MigrateJsResultVos task {} {}, cost {}ms", schemaKey, taskClient.getStatus(), (System.currentTimeMillis() - startTs));
             //成功
@@ -123,6 +129,9 @@ public List<MigrateJsResultVo> getJsResult(String jsNodeId, String virtualTarget
             }
         } catch (Exception e) {
             obsLogger.error("An error occurred while obtaining the results of model deduction", e);
+        } finally {
+            TaskClient<TaskDto> finalTaskClient = taskClient;
+            CommonUtils.ignoreAnyError(() -> Optional.ofNullable(finalTaskClient).ifPresent(TaskClient::close), TAG + "-closeTaskClient");
         }
         return new ArrayList<>();
     }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/error/TaskTargetProcessorExCode_15.java b/iengine/iengine-app/src/main/java/io/tapdata/error/TaskTargetProcessorExCode_15.java
index 831111fed7..59c9d81b8d 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/error/TaskTargetProcessorExCode_15.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/error/TaskTargetProcessorExCode_15.java
@@ -119,4 +119,9 @@ public interface TaskTargetProcessorExCode_15 {
 			describeCN = "创建索引时，检查索引是否存在失败"
 	)
 	String CREATE_INDEX_QUERY_EXISTS_INDEX_FAILED = "15021";
+	@TapExCode(
+			describe = "Failed to process events",
+			describeCN = "数据源处理失败"
+	)
+	String PROCESS_EVENTS_FAILED = "15022";
 }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
index 4ae58e67aa..5e74b53aea 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
@@ -213,7 +213,7 @@ public final void init(@NotNull Processor.Context context) throws Exception {
 				processorBaseContext.getTapTableMap().buildNodeName(processorBaseContext.getNode().getName());
 				processorBaseContext.getTapTableMap().preLoadSchema();
 			}
-			if (!getNode().disabledNode()) {
+			if (!getNode().disabledNode() || StringUtils.equalsAnyIgnoreCase(processorBaseContext.getTaskDto().getSyncType(),TaskDto.SYNC_TYPE_TEST_RUN,TaskDto.SYNC_TYPE_DEDUCE_SCHEMA)) {
 				doInit(context);
 			} else {
 				doInitWithDisableNode(context);
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/HazelcastSchemaTargetNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/HazelcastSchemaTargetNode.java
index 2d7f50613f..3359ac0a18 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/HazelcastSchemaTargetNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/HazelcastSchemaTargetNode.java
@@ -147,8 +147,8 @@ protected void doInit(@NotNull Context context) throws TapCodeException {
 							if (engine instanceof Closeable) {
 								((Closeable) engine).close();
 							}
-						} catch (Exception e) {
-							throw new CoreException("Resource cannot be released of {} Node, message: {}", isPythonNode ? "Python" : "JavaScript", e.getMessage());
+						} catch (Exception ignored) {
+							obsLogger.warn("Resource cannot be released of {} Node, message: {}", isPythonNode ? "Python" : "JavaScript", ignored.getMessage());
 						}
 					}
 				};
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 886d6d6588..153221d179 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
@@ -68,6 +68,7 @@
 import io.tapdata.flow.engine.V2.util.PdkUtil;
 import io.tapdata.flow.engine.V2.util.SyncTypeEnum;
 import io.tapdata.flow.engine.V2.util.TapEventUtil;
+import io.tapdata.inspect.AutoRecovery;
 import io.tapdata.node.pdk.ConnectorNodeService;
 import io.tapdata.pdk.apis.functions.PDKMethod;
 import io.tapdata.pdk.apis.functions.connection.GetTableNamesFunction;
@@ -182,6 +183,7 @@ private boolean needCdcDelay() {
 
 	@Override
 	protected void doInit(@NotNull Context context) throws TapCodeException {
+        AutoRecovery.setEnqueueConsumer(getNode().getTaskId(), this::enqueue);
 		if (needCdcDelay()) {
 			this.cdcDelayCalculation = new CdcDelay();
 		} else {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
index 668321f69d..74b1be8476 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
@@ -42,7 +42,6 @@
 import io.tapdata.entity.schema.value.TapMapValue;
 import io.tapdata.error.TapEventException;
 import io.tapdata.error.TapdataEventException;
-import io.tapdata.error.TaskProcessorExCode_11;
 import io.tapdata.error.TaskTargetProcessorExCode_15;
 import io.tapdata.exception.NodeException;
 import io.tapdata.exception.TapCodeException;
@@ -62,6 +61,7 @@
 import io.tapdata.flow.engine.V2.util.PdkUtil;
 import io.tapdata.flow.engine.V2.util.TapEventUtil;
 import io.tapdata.flow.engine.V2.util.TargetTapEventFilter;
+import io.tapdata.inspect.AutoRecovery;
 import io.tapdata.metric.collector.ISyncMetricCollector;
 import io.tapdata.milestone.MilestoneStage;
 import io.tapdata.milestone.MilestoneStatus;
@@ -182,7 +182,12 @@ protected void doInit(@NotNull Context context) throws TapCodeException {
 		everHandleTapTablePrimaryKeysMap = new ConcurrentHashMap<>();
 	}
 
-	@Override
+    @Override
+    protected boolean isRunning() {
+        return super.isRunning();
+    }
+
+    @Override
 	protected void doInitWithDisableNode(@NotNull Context context) throws TapCodeException {
 		queueConsumerThreadPool.submitSync(() -> {
 			super.doInitWithDisableNode(context);
@@ -207,7 +212,7 @@ protected void initExactlyOnceWriteIfNeed() {
 		if (null != dataProcessorContext.getTapTableMap()) {
 			dataProcessorContext.getTapTableMap().putNew(exactlyOnceTable.getId(), exactlyOnceTable, exactlyOnceTable.getId());
 		}
-		boolean create = createTable(exactlyOnceTable, new AtomicBoolean());
+		boolean create = createTable(exactlyOnceTable, new AtomicBoolean(),true);
 		if (create) {
 			obsLogger.info("Create exactly once write cache table: {}", exactlyOnceTable);
 			CreateIndexFunction createIndexFunction = connectorFunctions.getCreateIndexFunction();
@@ -259,7 +264,7 @@ protected void checkUnwindConfiguration(){
 		}
 	}
 
-	protected boolean createTable(TapTable tapTable, AtomicBoolean succeed) {
+	protected boolean createTable(TapTable tapTable, AtomicBoolean succeed,boolean init) {
 		if (getNode().disabledNode()) {
 			obsLogger.info("Target node has been disabled, task will skip: create table");
 			return false;
@@ -280,6 +285,7 @@ protected boolean createTable(TapTable tapTable, AtomicBoolean succeed) {
 				tapCreateTableEvent.set(createTableEvent(finalTapTable));
 				executeDataFuncAspect(CreateTableFuncAspect.class, () -> new CreateTableFuncAspect()
 						.createTableEvent(tapCreateTableEvent.get())
+						.setInit(init)
 						.connectorContext(getConnectorNode().getConnectorContext())
 						.dataProcessorContext(dataProcessorContext)
 						.start(), (createTableFuncAspect ->
@@ -592,116 +598,146 @@ protected void splitDDL2NewBatch(List<TapdataEvent> cdcEvents, Consumer<List<Tap
 		}
 	}
 
-	private void handleTapdataEvents(List<TapdataEvent> tapdataEvents) {
+	protected void handleTapdataEvents(List<TapdataEvent> tapdataEvents) {
 		AtomicReference<TapdataEvent> lastTapdataEvent = new AtomicReference<>();
 		List<TapEvent> tapEvents = new ArrayList<>();
 		List<TapRecordEvent> exactlyOnceWriteCache = new ArrayList<>();
 		List<TapdataShareLogEvent> tapdataShareLogEvents = new ArrayList<>();
-		if (null != getConnectorNode()) {
-			codecsFilterManager = getConnectorNode().getCodecsFilterManager();
-		}
+        Optional.ofNullable(getConnectorNode()).ifPresent(connectorNode -> codecsFilterManager = connectorNode.getCodecsFilterManager());
+
 		initAndGetExactlyOnceWriteLookupList();
-		boolean hasExactlyOnceWriteCache = false;
+        AtomicBoolean hasExactlyOnceWriteCache = new AtomicBoolean(false);
 		for (TapdataEvent tapdataEvent : tapdataEvents) {
-			if (!isRunning()) return;
-			try {
-				SyncStage syncStage = tapdataEvent.getSyncStage();
-				if (null != syncStage) {
-					if (syncStage == SyncStage.INITIAL_SYNC && firstBatchEvent.compareAndSet(false, true)) {
-                        syncMetricCollector.snapshotBegin();
-						executeAspect(new SnapshotWriteBeginAspect().dataProcessorContext(dataProcessorContext));
-					} else if (syncStage == SyncStage.CDC && firstStreamEvent.compareAndSet(false, true)) {
-                        syncMetricCollector.cdcBegin();
-						executeAspect(new CDCWriteBeginAspect().dataProcessorContext(dataProcessorContext));
-					}
-				}
-				if (tapdataEvent instanceof TapdataHeartbeatEvent) {
-					handleTapdataHeartbeatEvent(tapdataEvent);
-				} else if (tapdataEvent instanceof TapdataCompleteSnapshotEvent) {
-					handleTapdataCompleteSnapshotEvent();
-				} else if (tapdataEvent instanceof TapdataStartingCdcEvent) {
-					handleTapdataStartCdcEvent(tapdataEvent);
-				} else if (tapdataEvent instanceof TapdataStartedCdcEvent) {
-					flushShareCdcTableMetrics(tapdataEvent);
-				} else if (tapdataEvent instanceof TapdataTaskErrorEvent) {
-					throw ((TapdataTaskErrorEvent) tapdataEvent).getThrowable();
-				} else if (tapdataEvent instanceof TapdataShareLogEvent) {
-					handleTapdataShareLogEvent(tapdataShareLogEvents, tapdataEvent, lastTapdataEvent::set);
-				} else if (tapdataEvent instanceof TapdataCompleteTableSnapshotEvent) {
-					handleTapdataCompleteTableSnapshotEvent((TapdataCompleteTableSnapshotEvent) tapdataEvent);
-				} else if (tapdataEvent instanceof TapdataAdjustMemoryEvent) {
-					handleTapdataAdjustMemoryEvent((TapdataAdjustMemoryEvent) tapdataEvent);
-				} else {
-					if (tapdataEvent.isDML()) {
-						TapRecordEvent tapRecordEvent = handleTapdataRecordEvent(tapdataEvent);
-						if (null == tapRecordEvent) {
-							continue;
-						}
-						hasExactlyOnceWriteCache = handleExactlyOnceWriteCacheIfNeed(tapdataEvent, exactlyOnceWriteCache);
-						List<String> lookupTables = initAndGetExactlyOnceWriteLookupList();
-						String tgtTableNameFromTapEvent = getTgtTableNameFromTapEvent(tapRecordEvent);
-						if (null != lookupTables && lookupTables.contains(tgtTableNameFromTapEvent) && hasExactlyOnceWriteCache && eventExactlyOnceWriteCheckExists(tapdataEvent)) {
-							if (obsLogger.isDebugEnabled()) {
-								obsLogger.debug("Event check exactly once write exists, will ignore it: {}" + JSONUtil.obj2Json(tapRecordEvent));
-							}
-							continue;
-						} else {
-							if (SyncStage.CDC.equals(tapdataEvent.getSyncStage()) && null != lookupTables && lookupTables.contains(tgtTableNameFromTapEvent)) {
-								obsLogger.info("Target table {} stop look up exactly once cache", tgtTableNameFromTapEvent);
-								lookupTables.remove(tgtTableNameFromTapEvent);
-							}
-						}
-						tapEvents.add(tapRecordEvent);
-						if (null != tapdataEvent.getBatchOffset() || null != tapdataEvent.getStreamOffset()) {
-							lastTapdataEvent.set(tapdataEvent);
-						}
-					} else if (tapdataEvent.isDDL()) {
-						handleTapdataDDLEvent(tapdataEvent, tapEvents, lastTapdataEvent::set);
-					} else {
-						if (null != tapdataEvent.getTapEvent()) {
-							obsLogger.warn("Tap event type does not supported: " + tapdataEvent.getTapEvent().getClass() + ", will ignore it");
-						}
-					}
-				}
-			} catch (Throwable throwable) {
-				throw new TapdataEventException(TaskTargetProcessorExCode_15.HANDLE_EVENTS_FAILED, throwable).addEvent(tapdataEvent);
-			}
-		}
-		if (CollectionUtils.isNotEmpty(tapEvents)) {
-			try {
-				try {
-					if (checkExactlyOnceWriteEnableResult.getEnable() && hasExactlyOnceWriteCache) {
-						transactionBegin();
-					}
-					processEvents(tapEvents);
-					if (checkExactlyOnceWriteEnableResult.getEnable() && hasExactlyOnceWriteCache) {
-						processExactlyOnceWriteCache(tapdataEvents);
-						transactionCommit();
-					}
-				} catch (Exception e) {
-					if (checkExactlyOnceWriteEnableResult.getEnable() && hasExactlyOnceWriteCache) {
-						transactionRollback();
-					}
-					throw e;
-				}
-				flushOffsetByTapdataEventForNoConcurrent(lastTapdataEvent);
-			} catch (Throwable throwable) {
-				throw new RuntimeException(String.format("Process events failed: %s", throwable.getMessage()), throwable);
-			}
-		}
-		if (CollectionUtils.isNotEmpty(tapdataShareLogEvents)) {
-			try {
-				processShareLog(tapdataShareLogEvents);
-				flushOffsetByTapdataEventForNoConcurrent(lastTapdataEvent);
-			} catch (Throwable throwable) {
-				throw new RuntimeException(String.format("Process share log failed: %s", throwable.getMessage()), throwable);
-			}
+			if (!isRunning()) {
+                return;
+            }
+            handleTapdataEvent(tapEvents, tapdataShareLogEvents, lastTapdataEvent, hasExactlyOnceWriteCache, exactlyOnceWriteCache, tapdataEvent);
 		}
+
+        try {
+            processTapEvents(tapdataEvents, tapEvents, hasExactlyOnceWriteCache);
+
+            if (CollectionUtils.isNotEmpty(tapdataShareLogEvents)) {
+                processShareLog(tapdataShareLogEvents);
+            }
+
+            flushOffsetByTapdataEventForNoConcurrent(lastTapdataEvent);
+        } catch (Throwable throwable) {
+            throw new TapdataEventException(TaskTargetProcessorExCode_15.PROCESS_EVENTS_FAILED, throwable).addEvent(lastTapdataEvent.get());
+        }
+
 		if (firstStreamEvent.get()) {
 			executeAspect(new CDCHeartbeatWriteAspect().tapdataEvents(tapdataEvents).dataProcessorContext(dataProcessorContext));
 		}
 	}
 
+    protected void handleTapdataEvent(List<TapEvent> tapEvents, List<TapdataShareLogEvent> tapdataShareLogEvents
+        , AtomicReference<TapdataEvent> lastTapdataEvent, AtomicBoolean hasExactlyOnceWriteCache
+        , List<TapRecordEvent> exactlyOnceWriteCache, TapdataEvent tapdataEvent) {
+        try {
+            Optional.ofNullable(tapdataEvent.getSyncStage()).ifPresent(this::handleAspectWithSyncStage);
+
+            if (tapdataEvent instanceof TapdataHeartbeatEvent) {
+                handleTapdataHeartbeatEvent(tapdataEvent);
+            } else if (tapdataEvent instanceof TapdataCompleteSnapshotEvent) {
+                handleTapdataCompleteSnapshotEvent();
+            } else if (tapdataEvent instanceof TapdataStartingCdcEvent) {
+                handleTapdataStartCdcEvent(tapdataEvent);
+            } else if (tapdataEvent instanceof TapdataStartedCdcEvent) {
+                flushShareCdcTableMetrics(tapdataEvent);
+            } else if (tapdataEvent instanceof TapdataTaskErrorEvent) {
+                throw ((TapdataTaskErrorEvent) tapdataEvent).getThrowable();
+            } else if (tapdataEvent instanceof TapdataShareLogEvent) {
+                handleTapdataShareLogEvent(tapdataShareLogEvents, tapdataEvent, lastTapdataEvent::set);
+            } else if (tapdataEvent instanceof TapdataCompleteTableSnapshotEvent) {
+                handleTapdataCompleteTableSnapshotEvent((TapdataCompleteTableSnapshotEvent) tapdataEvent);
+            } else if (tapdataEvent instanceof TapdataAdjustMemoryEvent) {
+                handleTapdataAdjustMemoryEvent((TapdataAdjustMemoryEvent) tapdataEvent);
+            } else {
+                handleTapdataEvent(tapEvents, hasExactlyOnceWriteCache, exactlyOnceWriteCache, lastTapdataEvent, tapdataEvent);
+                if (tapdataEvent instanceof TapdataRecoveryEvent) {
+                    AutoRecovery.completed(getNode().getTaskId(), (TapdataRecoveryEvent) tapdataEvent);
+                }
+            }
+        } catch (Throwable throwable) {
+            throw new TapdataEventException(TaskTargetProcessorExCode_15.HANDLE_EVENTS_FAILED, throwable).addEvent(tapdataEvent);
+        }
+    }
+
+    protected void processTapEvents(List<TapdataEvent> tapdataEvents, List<TapEvent> tapEvents, AtomicBoolean hasExactlyOnceWriteCache) {
+        if (CollectionUtils.isEmpty(tapEvents)) return;
+
+        if (Boolean.TRUE.equals(checkExactlyOnceWriteEnableResult.getEnable()) && hasExactlyOnceWriteCache.get()) {
+            try {
+                transactionBegin();
+                processEvents(tapEvents);
+                processExactlyOnceWriteCache(tapdataEvents);
+                transactionCommit();
+            } catch (Exception e) {
+                transactionRollback();
+                throw e;
+            }
+        } else {
+            processEvents(tapEvents);
+        }
+    }
+
+    protected void handleAspectWithSyncStage(SyncStage syncStage) {
+        switch (syncStage) {
+            case INITIAL_SYNC:
+                if (firstBatchEvent.compareAndSet(false, true)) {
+                    syncMetricCollector.snapshotBegin();
+                    executeAspect(new SnapshotWriteBeginAspect().dataProcessorContext(dataProcessorContext));
+                }
+                break;
+            case CDC:
+                if (firstStreamEvent.compareAndSet(false, true)) {
+                    syncMetricCollector.cdcBegin();
+                    executeAspect(new CDCWriteBeginAspect().dataProcessorContext(dataProcessorContext));
+                }
+                break;
+            default:
+                break;
+        }
+    }
+
+    protected void handleTapdataEvent(List<TapEvent> tapEvents, AtomicBoolean hasExactlyOnceWriteCache, List<TapRecordEvent> exactlyOnceWriteCache, AtomicReference<TapdataEvent> lastTapdataEvent, TapdataEvent tapdataEvent) throws JsonProcessingException {
+        if (tapdataEvent.isDML()) {
+            handleTapdataEventDML(tapEvents, hasExactlyOnceWriteCache, exactlyOnceWriteCache, lastTapdataEvent, tapdataEvent);
+        } else if (tapdataEvent.isDDL()) {
+            handleTapdataDDLEvent(tapdataEvent, tapEvents, lastTapdataEvent::set);
+        } else {
+            if (null != tapdataEvent.getTapEvent()) {
+                obsLogger.warn("Tap event type does not supported: " + tapdataEvent.getTapEvent().getClass() + ", will ignore it");
+            }
+        }
+    }
+
+    protected void handleTapdataEventDML(List<TapEvent> tapEvents, AtomicBoolean hasExactlyOnceWriteCache, List<TapRecordEvent> exactlyOnceWriteCache, AtomicReference<TapdataEvent> lastTapdataEvent, TapdataEvent tapdataEvent) throws JsonProcessingException {
+        TapRecordEvent tapRecordEvent = handleTapdataRecordEvent(tapdataEvent);
+        if (null == tapRecordEvent) {
+            return;
+        }
+        hasExactlyOnceWriteCache.set(handleExactlyOnceWriteCacheIfNeed(tapdataEvent, exactlyOnceWriteCache));
+        List<String> lookupTables = initAndGetExactlyOnceWriteLookupList();
+        String tgtTableNameFromTapEvent = getTgtTableNameFromTapEvent(tapRecordEvent);
+        if (null != lookupTables && lookupTables.contains(tgtTableNameFromTapEvent) && hasExactlyOnceWriteCache.get() && eventExactlyOnceWriteCheckExists(tapdataEvent)) {
+            if (obsLogger.isDebugEnabled()) {
+                obsLogger.debug("Event check exactly once write exists, will ignore it: {}" + JSONUtil.obj2Json(tapRecordEvent));
+            }
+            return;
+        } else {
+            if (SyncStage.CDC.equals(tapdataEvent.getSyncStage()) && null != lookupTables && lookupTables.contains(tgtTableNameFromTapEvent)) {
+                obsLogger.info("Target table {} stop look up exactly once cache", tgtTableNameFromTapEvent);
+                lookupTables.remove(tgtTableNameFromTapEvent);
+            }
+        }
+        tapEvents.add(tapRecordEvent);
+        if (null != tapdataEvent.getBatchOffset() || null != tapdataEvent.getStreamOffset()) {
+            lastTapdataEvent.set(tapdataEvent);
+        }
+    }
+
 	private void handleTapdataAdjustMemoryEvent(TapdataAdjustMemoryEvent tapdataEvent) {
 		try {
 			int mode = tapdataEvent.getMode();
@@ -1344,7 +1380,7 @@ private boolean tableEnableExactlyOnceWrite(SyncStage syncStage, String tableId)
 		return SyncStage.CDC.equals(syncStage) && exactlyOnceWriteTables.contains(tableId);
 	}
 
-	private List<String> initAndGetExactlyOnceWriteLookupList() {
+	protected List<String> initAndGetExactlyOnceWriteLookupList() {
 		return exactlyOnceWriteNeedLookupTables.computeIfAbsent(Thread.currentThread().getName(), k -> {
 			Node node = getNode();
 			if (node instanceof TableNode) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
index d3f20f8704..61b0cbd01a 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
@@ -146,7 +146,7 @@ private void initTargetDB() {
 					if (!isRunning()) {
 						return;
 					}
-					createTable(tapTableMap, funcAspect, node, existsDataProcessEnum, tableId);
+					createTable(tapTableMap, funcAspect, node, existsDataProcessEnum, tableId,true);
 				}
 			}
 
@@ -174,7 +174,7 @@ private void initTargetDB() {
 										if (!isRunning()) {
 											return;
 										}
-										createTable(tapTableMap, funcAspect, node, existsDataProcessEnum, tableId);
+										createTable(tapTableMap, funcAspect, node, existsDataProcessEnum, tableId,true);
 									}
 								}
 							}
@@ -185,7 +185,7 @@ private void initTargetDB() {
 		}));
 	}
 
-	protected void createTable(TapTableMap<String, TapTable> tapTableMap, TableInitFuncAspect funcAspect, Node<?> node, ExistsDataProcessEnum existsDataProcessEnum, String tableId) {
+	protected void createTable(TapTableMap<String, TapTable> tapTableMap, TableInitFuncAspect funcAspect, Node<?> node, ExistsDataProcessEnum existsDataProcessEnum, String tableId,boolean init) {
 		TapTable tapTable = tapTableMap.get(tableId);
 		List<String> updateConditionFields = getUpdateConditionFields(node, tapTable);
 		if (null == tapTable) {
@@ -196,11 +196,11 @@ protected void createTable(TapTableMap<String, TapTable> tapTableMap, TableInitF
 		if (StringUtils.isNotBlank(tableId) && StringUtils.equalsAny(tableId, ExactlyOnceUtil.EXACTLY_ONCE_CACHE_TABLE_NAME)) {
 			return;
 		}
-		dropTable(existsDataProcessEnum, tableId);
+		dropTable(existsDataProcessEnum, tableId, init);
 //		boolean createUnique = tapTable.getIndexList() != null && tapTable.getIndexList().stream().anyMatch(idx -> !idx.isPrimary() && idx.isUnique() && (idx.getIndexFields().size() == updateConditionFields.size()) &&
 //				(idx.getIndexFields().stream().allMatch(idxField -> updateConditionFields.contains(idxField.getName()))));
 		AtomicBoolean succeed = new AtomicBoolean(false);
-		boolean createdTable = createTable(tapTable, succeed);
+		boolean createdTable = createTable(tapTable, succeed, init);
 		clearData(existsDataProcessEnum, tableId);
 //		createUnique &= succeed.get();
 		createTargetIndex(updateConditionFields, succeed.get(), tableId, tapTable, createdTable);
@@ -438,7 +438,7 @@ protected void clearData(ExistsDataProcessEnum existsDataProcessEnum, String tab
 		}
 	}
 
-	protected void dropTable(ExistsDataProcessEnum existsDataProcessEnum, String tableId) {
+	protected void dropTable(ExistsDataProcessEnum existsDataProcessEnum, String tableId, boolean init) {
 		if (SyncTypeEnum.CDC == syncType || existsDataProcessEnum != ExistsDataProcessEnum.DROP_TABLE) return;
 
 		AtomicReference<TapDropTableEvent> tapDropTableEvent = new AtomicReference<>();
@@ -447,6 +447,7 @@ protected void dropTable(ExistsDataProcessEnum existsDataProcessEnum, String tab
 			if (dropTableFunction != null) {
 				tapDropTableEvent.set(dropTableEvent(tableId));
 				executeDataFuncAspect(DropTableFuncAspect.class, () -> new DropTableFuncAspect()
+						.setInit(init)
 						.dropTableEvent(tapDropTableEvent.get())
 						.connectorContext(getConnectorNode().getConnectorContext())
 						.dataProcessorContext(dataProcessorContext)
@@ -726,7 +727,7 @@ protected boolean executeDropFieldFunction(TapDropFieldEvent tapDropFieldEvent)
 	private boolean executeCreateTableFunction(TapCreateTableEvent tapCreateTableEvent) {
 		String tgtTableName = getTgtTableNameFromTapEvent(tapCreateTableEvent);
 		TapTable tgtTapTable = dataProcessorContext.getTapTableMap().get(tgtTableName);
-		return createTable(tgtTapTable, new AtomicBoolean());
+		return createTable(tgtTapTable, new AtomicBoolean(),false);
 	}
 
 	protected boolean executeCreateIndexFunction(TapCreateIndexEvent tapCreateIndexEvent) {
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 1dee58ef5d..61ee21ac04 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
@@ -25,10 +25,7 @@
 import io.tapdata.entity.schema.TapIndexField;
 import io.tapdata.entity.schema.TapTable;
 import io.tapdata.entity.schema.type.TapString;
-import io.tapdata.entity.schema.value.DateTime;
-import io.tapdata.entity.schema.value.TapArrayValue;
-import io.tapdata.entity.schema.value.TapMapValue;
-import io.tapdata.entity.schema.value.TapStringValue;
+import io.tapdata.entity.schema.value.*;
 import io.tapdata.entity.utils.DataMap;
 import io.tapdata.error.TapEventException;
 import io.tapdata.error.TaskMergeProcessorExCode_16;
@@ -844,7 +841,7 @@ protected void initSourceConnectionMap(List<MergeTableProperties> mergeTableProp
 		}
 	}
 
-	private void initSourcePkOrUniqueFieldMap(List<MergeTableProperties> mergeTableProperties) {
+	protected void initSourcePkOrUniqueFieldMap(List<MergeTableProperties> mergeTableProperties) {
 		if (null == mergeTableProperties) {
 			this.sourcePkOrUniqueFieldMap = new HashMap<>();
 			Node<?> node = this.processorBaseContext.getNode();
@@ -858,32 +855,34 @@ private void initSourcePkOrUniqueFieldMap(List<MergeTableProperties> mergeTableP
 			if (null == preNode) {
 				throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NODE_NOT_FOUND, String.format("- Node ID: %s", sourceNodeId));
 			}
-			String nodeName = preNode.getName();
-			String tableName = getTableName(preNode);
-			TapTable tapTable = tapTableMap.get(tableName);
-			MergeTableProperties.MergeType mergeType = mergeProperty.getMergeType();
-			List<String> arrayKeys = mergeProperty.getArrayKeys();
-			Collection<String> primaryKeys = tapTable.primaryKeys(true);
-			List<String> fieldNames;
-			switch (mergeType) {
-				case appendWrite:
-				case updateOrInsert:
-				case updateWrite:
-					if (CollectionUtils.isEmpty(primaryKeys)) {
-						throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_PRIMARY_KEY, String.format("- Table name: %s\n- Node name: %s\n- Merge operation: %s", tableName, nodeName, mergeType));
-					}
-					fieldNames = new ArrayList<>(primaryKeys);
-					break;
-				case updateIntoArray:
-					if (CollectionUtils.isEmpty(arrayKeys)) {
-						throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_ARRAY_KEY, String.format("- Table name: %s- Node name: %s\n", tableName, nodeName));
-					}
-					fieldNames = arrayKeys;
-					break;
-				default:
-					throw new RuntimeException("Unrecognized merge type: " + mergeType);
+			if(!preNode.disabledNode()){
+				String nodeName = preNode.getName();
+				String tableName = getTableName(preNode);
+				TapTable tapTable = tapTableMap.get(tableName);
+				MergeTableProperties.MergeType mergeType = mergeProperty.getMergeType();
+				List<String> arrayKeys = mergeProperty.getArrayKeys();
+				Collection<String> primaryKeys = tapTable.primaryKeys(true);
+				List<String> fieldNames;
+				switch (mergeType) {
+					case appendWrite:
+					case updateOrInsert:
+					case updateWrite:
+						if (CollectionUtils.isEmpty(primaryKeys)) {
+							throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_PRIMARY_KEY, String.format("- Table name: %s\n- Node name: %s\n- Merge operation: %s", tableName, nodeName, mergeType));
+						}
+						fieldNames = new ArrayList<>(primaryKeys);
+						break;
+					case updateIntoArray:
+						if (CollectionUtils.isEmpty(arrayKeys)) {
+							throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_ARRAY_KEY, String.format("- Table name: %s- Node name: %s\n", tableName, nodeName));
+						}
+						fieldNames = arrayKeys;
+						break;
+					default:
+						throw new RuntimeException("Unrecognized merge type: " + mergeType);
+				}
+				this.sourcePkOrUniqueFieldMap.put(sourceNodeId, fieldNames);
 			}
-			this.sourcePkOrUniqueFieldMap.put(sourceNodeId, fieldNames);
 			initSourcePkOrUniqueFieldMap(mergeProperty.getChildren());
 		}
 	}
@@ -1001,7 +1000,7 @@ protected boolean needLookup(TapdataEvent tapdataEvent) {
 		SyncStage syncStage = tapdataEvent.getSyncStage();
 		if (isInvalidOperation(tapdataEvent)) return false;
 		String op = getOp(tapdataEvent);
-		if (op.equals(OperationType.DELETE.getOp())) {
+		if (op.equals(OperationType.DELETE.getOp()) || op.equals(OperationType.UPDATE.getOp())) {
 			return false;
 		}
 		String preNodeId = getPreNodeId(tapdataEvent);
@@ -1234,7 +1233,7 @@ protected void upsertCache(List<TapdataEvent> tapdataEvents, MergeTablePropertie
 	protected void transformDateTime(Map<String, Object> after) {
 		mapIterator.iterate(after, (key, value, recursive) -> {
 			if (value instanceof DateTime) {
-				return ((DateTime) value).toDate();
+				return ((DateTime) value).toInstant();
 			}
 			return value;
 		});
@@ -1569,7 +1568,14 @@ private Map<String, Object> mockLookupMap(MergeTableProperties childMergePropert
 		String tableName = getTableName(getPreNode(childMergeProperty.getId()));
 		TapTable tapTable = processorBaseContext.getTapTableMap().get(tableName);
 		LinkedHashMap<String, TapField> nameFieldMap = tapTable.getNameFieldMap();
-		nameFieldMap.keySet().forEach(key -> lookupMap.put(key, null));
+		while (isRunning()) {
+			try {
+				nameFieldMap.keySet().forEach(key -> lookupMap.put(key, null));
+				break;
+			} catch (ConcurrentModificationException ignored) {
+				// nothing to do
+			}
+		}
 		return lookupMap;
 	}
 
@@ -2148,8 +2154,8 @@ protected void recursiveMergeInfoTransformToTapValue(List<MergeLookupResult> mer
 						return new TapStringValue(value.toString())
 								.tapType(new TapString(24L, true))
 								.originType(BsonType.OBJECT_ID.name());
-					} else if (value instanceof Document) {
-						return new TapMapValue((Document) value);
+					} else if (value instanceof Map) {
+						return new TapMapValue((Map<String, Object>) value);
 					} else if (value instanceof List) {
 						return new TapArrayValue((List<Object>) value);
 					}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/join/HazelcastJoinProcessor.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/join/HazelcastJoinProcessor.java
index bae136a4f8..82f26718f7 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/join/HazelcastJoinProcessor.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/join/HazelcastJoinProcessor.java
@@ -12,6 +12,9 @@
 import com.tapdata.tm.commons.dag.process.JoinProcessorNode;
 import com.tapdata.tm.commons.externalStorage.ExternalStorageDto;
 import io.tapdata.construct.constructImpl.BytesIMap;
+import io.tapdata.entity.codec.filter.EntryFilter;
+import io.tapdata.entity.codec.filter.MapIteratorEx;
+import io.tapdata.entity.codec.filter.impl.AllLayerMapIterator;
 import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapRecordEvent;
@@ -19,6 +22,7 @@
 import io.tapdata.entity.schema.TapField;
 import io.tapdata.entity.schema.TapIndex;
 import io.tapdata.entity.schema.TapTable;
+import io.tapdata.entity.schema.value.DateTime;
 import io.tapdata.error.TaskProcessorExCode_11;
 import io.tapdata.exception.NodeException;
 import io.tapdata.exception.TapCodeException;
@@ -36,15 +40,7 @@
 import org.jetbrains.annotations.NotNull;
 
 import javax.annotation.Nonnull;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 
@@ -79,6 +75,7 @@ public class HazelcastJoinProcessor extends HazelcastProcessorBaseNode {
 	private List<String> leftPrimaryKeys;
 	private List<String> rightPrimaryKeys;
 	private TapTable tapTable;
+	private MapIteratorEx mapIterator;
 
 	public HazelcastJoinProcessor(ProcessorBaseContext processorBaseContext) {
 		super(processorBaseContext);
@@ -106,6 +103,7 @@ private void pkChecker() {
 	public void doInit(@Nonnull Context context) throws TapCodeException {
 		super.doInit(context);
 		this.context = context;
+		this.mapIterator = new AllLayerMapIterator();
 		initNode();
 	}
 
@@ -151,7 +149,7 @@ public static void clearCache(Node<?> node) {
 		});
 	}
 
-	private void initNode() throws TapCodeException {
+	public void initNode() throws TapCodeException {
 		Node<?> node = processorBaseContext.getNode();
 		if (verifyJoinNode(node)) {
 			vatidate(node);
@@ -269,7 +267,7 @@ protected void tryProcess(TapdataEvent tapdataEvent, BiConsumer<TapdataEvent, Pr
 			consumer.accept(tapdataEvent, null);
 			return;
 		}
-
+		transformDateTime(before,after);
 		List<JoinResult> joinResults;
 		if (tapdataEvent.getNodeIds().contains(leftNodeId)) {
 			joinResults = leftJoinLeftProcess(before, after, opType);
@@ -740,6 +738,17 @@ public static String project(Map<String, Object> record, List<String> fields) {
 		return Arrays.deepToString(key);
 	}
 
+	public void transformDateTime(Map<String, Object> before,Map<String, Object> after) {
+		EntryFilter entryFilter = (key, value, recursive) -> {
+			if (value instanceof DateTime) {
+				return ((DateTime) value).toDate();
+			}
+			return value;
+		};
+		mapIterator.iterate(before,entryFilter);
+		mapIterator.iterate(after, entryFilter);
+	}
+
 	public class JoinResult {
 		private Map<String, Object> before;
 		private Map<String, Object> after;
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
index f89dd753a3..b25043792c 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
@@ -648,7 +648,6 @@ protected void stopTask(String taskId) {
 		taskDtoTaskClient.terminalMode(TerminalMode.STOP_GRACEFUL);
 		taskDtoTaskClient.getTask().setSnapShotInterrupt(true);
 		stopTaskAndClear(taskDtoTaskClient, StopTaskResource.STOPPED, taskId);
-		AspectUtils.executeAspect(new TaskStopAspect().task(taskDtoTaskClient.getTask()));
 	}
 
 	private void stopTaskAndClear(TaskClient<TaskDto> taskDtoTaskClient, StopTaskResource stopped, String taskId) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
index acb0987bf5..8901374b86 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
@@ -65,4 +65,8 @@ default int getRetryTime() {
 	default boolean resume() {
 		throw new UnsupportedOperationException();
 	}
+
+	default void close() {
+		throw new UnsupportedOperationException();
+	}
 }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
index 6752827355..cd322f3be1 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
@@ -19,6 +19,7 @@
 import io.tapdata.flow.engine.V2.task.TerminalMode;
 import io.tapdata.flow.engine.V2.util.ConsumerImpl;
 import io.tapdata.flow.engine.V2.util.SupplierImpl;
+import io.tapdata.inspect.AutoRecovery;
 import io.tapdata.observable.logging.ObsLogger;
 import io.tapdata.observable.logging.ObsLoggerFactory;
 import io.tapdata.pdk.core.utils.CommonUtils;
@@ -59,6 +60,7 @@ public class HazelcastTaskClient implements TaskClient<TaskDto> {
 	private long lastRetryTimeMillis;
 	private final AtomicInteger retryCounter;
 	private AtomicBoolean retrying;
+	private final AutoRecovery autoRecovery;
 
 	public static HazelcastTaskClient create(TaskDto taskDto, ClientMongoOperator clientMongoOperator, ConfigurationCenter configurationCenter, HazelcastInstance hazelcastInstance) {
 		return new HazelcastTaskClient(null, taskDto, clientMongoOperator, configurationCenter, hazelcastInstance);
@@ -83,6 +85,7 @@ public HazelcastTaskClient(Job job, TaskDto taskDto, ClientMongoOperator clientM
 		cacheNode.ifPresent(c -> cacheName = ((CacheNode) c).getCacheName());
 		this.retryCounter = new AtomicInteger(0);
 		this.retrying = new AtomicBoolean(false);
+        this.autoRecovery = AutoRecovery.init(taskDto.getId().toHexString());
 	}
 
 	@Override
@@ -162,17 +165,27 @@ public synchronized boolean stop() {
 		return false;
 	}
 
+	@Override
 	public void close() {
 		ObsLogger obsLogger = ObsLoggerFactory.getInstance().getObsLogger(taskDto);
 		CommonUtils.handleAnyError(
 				() -> {
-					monitorManager.close();
+					if (monitorManager != null) monitorManager.close();
 					obsLogger.info(String.format("Closed task monitor(s)\n%s", monitorManager));
 				},
 				err -> {
 					obsLogger.warn(String.format("Close task monitor(s) failed, error: %s\n  %s", err.getMessage(), Log4jUtil.getStackString(err)));
 				}
 		);
+            CommonUtils.handleAnyError(
+                () -> {
+					if(null != autoRecovery) autoRecovery.close();
+                    obsLogger.info(String.format("Closed task auto recovery instance\n  %s", autoRecovery));
+                },
+                err -> {
+                    obsLogger.warn(String.format("Closed task auto recovery instance failed, error: %s\n  %s", err.getMessage(), Log4jUtil.getStackString(err)));
+                }
+            );
 		CommonUtils.handleAnyError(
 				() -> {
 					AspectUtils.executeAspect(new TaskStopAspect().task(taskDto).error(error));
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 bc8ebc1f0b..96f9408275 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
@@ -42,11 +42,7 @@
 import com.tapdata.tm.commons.dag.nodes.CacheNode;
 import com.tapdata.tm.commons.dag.nodes.DataParentNode;
 import com.tapdata.tm.commons.dag.nodes.TableNode;
-import com.tapdata.tm.commons.dag.process.MergeTableNode;
-import com.tapdata.tm.commons.dag.process.MigrateDateProcessorNode;
-import com.tapdata.tm.commons.dag.process.MigrateFieldRenameProcessorNode;
-import com.tapdata.tm.commons.dag.process.ProcessorNode;
-import com.tapdata.tm.commons.dag.process.TableRenameProcessNode;
+import com.tapdata.tm.commons.dag.process.*;
 import com.tapdata.tm.commons.dag.vo.ReadPartitionOptions;
 import com.tapdata.tm.commons.schema.*;
 import com.tapdata.tm.commons.task.dto.ErrorEvent;
@@ -459,7 +455,8 @@ public static HazelcastBaseNode createNode(
 			TaskConfig taskConfig
 	) throws Exception {
 		List<RelateDataBaseTable> nodeSchemas = new ArrayList<>();
-		if ((node instanceof ProcessorNode || node instanceof MigrateDateProcessorNode) && node.disabledNode()) {
+		if (!StringUtils.equalsAnyIgnoreCase(taskDto.getSyncType(),TaskDto.SYNC_TYPE_TEST_RUN,TaskDto.SYNC_TYPE_DEDUCE_SCHEMA) &&
+				(node instanceof ProcessorNode || node instanceof MigrateProcessorNode) && node.disabledNode()) {
 			HazelcastBlank newNode = new HazelcastBlank(
 					DataProcessorContext.newBuilder()
 							.withTaskDto(taskDto)
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/inspect/InspectFactory.java b/iengine/iengine-app/src/main/java/io/tapdata/inspect/InspectFactory.java
index 898dce4191..29662272cf 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/inspect/InspectFactory.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/inspect/InspectFactory.java
@@ -27,25 +27,26 @@ public class InspectFactory {
 //                .setClientMongoOperator(clientMongoOperator)
 //                .setSettingService(settingService);
 //    }
-    private static InspectService inspectService;
+    private static volatile InspectService inspectService;
 
 
     public static InspectService getInstance(ClientMongoOperator clientMongoOperator, SettingService settingService) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
         if (inspectService == null) {
             synchronized (InspectFactory.class) {
                 if (inspectService == null) {
+                    InspectService inspectServicePrivate;
                     io.tapdata.factory.InspectType inspectType = InspectType.INSPECT_SERVICE_IMPL;
                     String clazz = inspectType.getClazz();
                     Constructor<?> declaredConstructor = Class.forName(clazz).getDeclaredConstructor(inspectType.getClasses());
                     declaredConstructor.setAccessible(true);
                     Object instance = declaredConstructor.newInstance();
                     if (instance instanceof InspectService) {
-                        inspectService = (InspectService) instance;
+                        inspectServicePrivate = (InspectService) instance;
                     } else {
                         throw new IllegalArgumentException("Implementation class must be InspectImpl: " + clazz);
                     }
-                    inspectService.init(clientMongoOperator,settingService);
-                    return inspectService;
+                    inspectServicePrivate.init(clientMongoOperator,settingService);
+                    inspectService = inspectServicePrivate;
                 }
             }
         }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/services/JSProcessNodeTestRunService.java b/iengine/iengine-app/src/main/java/io/tapdata/services/JSProcessNodeTestRunService.java
index 76bf9e8188..28e07a40d1 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/services/JSProcessNodeTestRunService.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/services/JSProcessNodeTestRunService.java
@@ -15,6 +15,7 @@
 import io.tapdata.observable.logging.ObsLoggerFactory;
 import io.tapdata.observable.logging.appender.ScriptNodeProcessNodeAppender;
 import io.tapdata.observable.logging.with.FixedSizeBlockingDeque;
+import io.tapdata.pdk.core.utils.CommonUtils;
 import io.tapdata.service.skeleton.annotation.RemoteService;
 import io.tapdata.websocket.handler.TestRunTaskHandler;
 
@@ -30,6 +31,7 @@
 @RemoteService
 public class JSProcessNodeTestRunService {
 
+    public static final String TAG = JSProcessNodeTestRunService.class.getSimpleName();
     private final Map<String, TestRunEntity> taskDtoMap = new ConcurrentHashMap<>();
     public static final long cache_time = 3000L;
 
@@ -99,6 +101,8 @@ public Object testRun(Map<String, Object> events, final String nodeId, final int
             resultMap.put("message", throwable.getMessage());
         } finally {
             ObsLoggerFactory.getInstance().removeTaskLogger(taskDto);
+            TaskClient<TaskDto> finalTaskClient = taskClient;
+            CommonUtils.ignoreAnyError(() -> Optional.ofNullable(finalTaskClient).ifPresent(TaskClient::close), TAG + "-closeTaskClient");
             taskDtoMap.remove(taskId);
             logger.info("test run task {} {}, cost {}ms", taskId, null == taskClient ? "error" : taskClient.getStatus(), (System.currentTimeMillis() - startTs));
         }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/task/TapdataTaskScheduler.java b/iengine/iengine-app/src/main/java/io/tapdata/task/TapdataTaskScheduler.java
index 977123f56d..1475ff2333 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/task/TapdataTaskScheduler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/task/TapdataTaskScheduler.java
@@ -269,18 +269,4 @@ private void stopTaskFromScheduler(String task_id) {
 		taskExecuteInfo.getScheduleTask().setStatus("paused");
 	}
 
-	public static void main(String[] args) throws InterruptedException {
-		ScheduledExecutorService executorService = Executors.newScheduledThreadPool(20);
-		ConcurrentTaskScheduler taskScheduler = new ConcurrentTaskScheduler(executorService);
-//        PeriodicTrigger trigger = new PeriodicTrigger(5000);
-		Future<?> schedule = taskScheduler.submit(() -> {
-			System.out.println("Already done.");
-		});
-
-		while (true) {
-			System.out.println(schedule.isDone());
-			Thread.sleep(1000);
-		}
-	}
-
 }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/TestRunTaskHandler.java b/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/TestRunTaskHandler.java
index 96181aed1a..4a2b5cf0ba 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/TestRunTaskHandler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/TestRunTaskHandler.java
@@ -11,15 +11,18 @@
 import io.tapdata.flow.engine.V2.task.TaskService;
 import io.tapdata.observable.logging.ObsLogger;
 import io.tapdata.observable.logging.ObsLoggerFactory;
+import io.tapdata.pdk.core.utils.CommonUtils;
 import io.tapdata.websocket.EventHandlerAnnotation;
 import io.tapdata.websocket.WebSocketEventHandler;
 import io.tapdata.websocket.WebSocketEventResult;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
 @EventHandlerAnnotation(type = "testRun")
 public class TestRunTaskHandler implements WebSocketEventHandler<WebSocketEventResult> {
+	public static final String TAG = TestRunTaskHandler.class.getSimpleName();
 	private ClientMongoOperator clientMongoOperator;
 
 	private TaskService<TaskDto> taskService;
@@ -67,6 +70,8 @@ public WebSocketEventResult handle(Map event) {
 			}
 			return WebSocketEventResult.handleFailed(WebSocketEventResult.Type.TEST_RUN, throwable.getMessage());
 		} finally {
+			TaskClient<TaskDto> finalTaskClient = taskClient;
+			CommonUtils.ignoreAnyError(() -> Optional.ofNullable(finalTaskClient).ifPresent(TaskClient::close), TAG + "-closeTaskClient");
 			taskDtoMap.remove(taskId);
 		}
 
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/TapdataRecoveryEvent.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/TapdataRecoveryEvent.java
new file mode 100644
index 0000000000..ddaec171ac
--- /dev/null
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/TapdataRecoveryEvent.java
@@ -0,0 +1,64 @@
+package com.tapdata.entity;
+
+import io.tapdata.entity.event.TapEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.Map;
+
+/**
+ * 修复事件
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/7 11:01 Create
+ */
+@Setter
+@Getter
+public class TapdataRecoveryEvent extends TapdataEvent {
+
+    public static final String EVENT_INFO_AUTO_RECOVERY_TASK = "AUTO_RECOVERY_TASK";
+
+    public static final String RECOVERY_TYPE_BEGIN = "BEGIN";
+    public static final String RECOVERY_TYPE_DATA = "DATA";
+    public static final String RECOVERY_TYPE_END = "END";
+
+    private String inspectTaskId;
+    private String recoveryType;
+
+    public TapdataRecoveryEvent() {
+    }
+
+    public TapdataRecoveryEvent(String inspectTaskId, String recoveryType) {
+        this.inspectTaskId = inspectTaskId;
+        this.recoveryType = recoveryType;
+    }
+
+    @Override
+    public boolean isConcurrentWrite() {
+        return RECOVERY_TYPE_DATA.equals(recoveryType);
+    }
+
+    public static TapdataRecoveryEvent createBegin(String inspectTaskId) {
+        return new TapdataRecoveryEvent(inspectTaskId, RECOVERY_TYPE_BEGIN);
+    }
+
+    public static TapdataRecoveryEvent createInsert(String inspectTaskId, String tableId, Map<String, Object> after) {
+        TapdataRecoveryEvent tapdataEvent = new TapdataRecoveryEvent(inspectTaskId, RECOVERY_TYPE_DATA);
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().after(after).table(tableId);
+        insertRecordEvent.addInfo(EVENT_INFO_AUTO_RECOVERY_TASK, inspectTaskId);
+        tapdataEvent.setTapEvent(insertRecordEvent);
+        return tapdataEvent;
+    }
+
+    public static TapdataRecoveryEvent createEnd(String inspectTaskId) {
+        return new TapdataRecoveryEvent(inspectTaskId, RECOVERY_TYPE_END);
+    }
+
+    public static boolean isRecoveryEvent(TapEvent tapEvent) {
+        if (null != tapEvent.getInfo()) {
+            return tapEvent.getInfo().containsKey(EVENT_INFO_AUTO_RECOVERY_TASK);
+        }
+        return false;
+    }
+}
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/Inspect.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/Inspect.java
index 9f851f2c21..82f15ab9ad 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/Inspect.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/Inspect.java
@@ -35,6 +35,7 @@ public class Inspect extends BaseEntity implements Serializable {
 	 * 差异结果模式：All,OnSourceExists
 	 */
 	private String inspectDifferenceMode;
+	private Boolean enableRecovery;
 
 	private InspectTiming timing;
 	private InspectLimit limit;
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/InspectStatus.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/InspectStatus.java
index 844cae829e..d0ade331e2 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/InspectStatus.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/inspect/InspectStatus.java
@@ -10,6 +10,7 @@ public enum InspectStatus {
 	PAUSE("pause"),
 	SCHEDULING("scheduling"),
 	RUNNING("running"),
+	RECOVERING("recovering"),
 	ERROR("error"),
 	DONE("done"),
 	WAITING("waiting"),
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/FieldProcessUtil.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/FieldProcessUtil.java
index 0732065966..dff4495753 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/FieldProcessUtil.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/FieldProcessUtil.java
@@ -564,11 +564,19 @@ public static Date convert2Date(Object value) {
 		return result;
 	}
 
-	private static  <T> T convertNumber(Object value, Function<String, T> fn) {
+	protected static  <T> T convertNumber(Object value, Function<String, T> fn) {
+		value = handleDateTime(value);
 		String str = String.valueOf(value);
 		if (StringUtils.isBlank(str)) {
 			return null;
 		}
 		return fn.apply(str);
 	}
+
+	protected static Object handleDateTime(Object value) {
+		if (value instanceof DateTime) {
+			value = ((DateTime) value).toEpochMilli();
+		}
+		return value;
+	}
 }
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/util/CustomTcp.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/util/CustomTcp.java
index c7982c1703..36c2ca140e 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/util/CustomTcp.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/util/CustomTcp.java
@@ -18,6 +18,8 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -357,9 +359,10 @@ public static byte[] bufBytes(byte[] begin, byte[] after) {
 	 * @author lidi
 	 * @date 2021/5/18 11:49
 	 */
-	private static String getStringRandom(int length) {
+	private static Random random = new Random();
+
+    private static String getStringRandom(int length) {
 		String val = "";
-		Random random = new Random();
 		//参数length，表示生成几位随机数
 		for (int i = 0; i < length; i++) {
 			String charOrNum = random.nextInt(2) % 2 == 0 ? "char" : "num";
diff --git a/iengine/iengine-common/src/main/java/io/tapdata/error/TaskInspectExCode_27.java b/iengine/iengine-common/src/main/java/io/tapdata/error/TaskInspectExCode_27.java
index 49cafd13b2..5d757680ac 100644
--- a/iengine/iengine-common/src/main/java/io/tapdata/error/TaskInspectExCode_27.java
+++ b/iengine/iengine-common/src/main/java/io/tapdata/error/TaskInspectExCode_27.java
@@ -24,4 +24,19 @@ public interface TaskInspectExCode_27 {
 
 	@TapExCode
 	String COMMAND_COUNT_PARAM_ERROR = "27005";
+
+	@TapExCode
+	String AUTO_RECOVERY_DUPLICATE = "27006";
+
+	@TapExCode
+	String AUTO_RECOVERY_NOT_EXISTS = "27007";
+
+    @TapExCode
+    String AUTO_RECOVERY_CLIENT_DUPLICATE = "27008";
+
+	@TapExCode
+	String AUTO_RECOVERY_TIMEOUT = "27009";
+
+	@TapExCode
+	String AUTO_RECOVERY_NOT_FOUND_RECOVERY_TYPE = "27010";
 }
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/ObservableAspectTask.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/ObservableAspectTask.java
index ca510a955c..daf02c330a 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/ObservableAspectTask.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/ObservableAspectTask.java
@@ -1,6 +1,7 @@
 package io.tapdata.observable.metric;
 
 import com.google.common.collect.HashBiMap;
+import com.tapdata.constant.Log4jUtil;
 import com.tapdata.tm.commons.dag.Node;
 import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
 import com.tapdata.tm.commons.task.dto.TaskDto;
@@ -12,27 +13,37 @@
 import io.tapdata.aspect.taskmilestones.SnapshotWriteTableCompleteAspect;
 import io.tapdata.entity.aspect.Aspect;
 import io.tapdata.entity.aspect.AspectInterceptResult;
+import io.tapdata.entity.logger.TapLogger;
 import io.tapdata.entity.simplify.pretty.ClassHandlers;
 import io.tapdata.entity.utils.InstanceFactory;
 import io.tapdata.module.api.PipelineDelay;
 import io.tapdata.observable.metric.handler.*;
 import io.tapdata.observable.metric.util.SyncGetMemorySizeHandler;
+import io.tapdata.observable.metric.util.TapCompletableFutureEx;
+import io.tapdata.observable.metric.util.TapCompletableFutureTaskEx;
+import io.tapdata.pdk.core.utils.CommonUtils;
 
 import java.math.BigDecimal;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
 @AspectTaskSession(includeTypes = {TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC, TaskDto.SYNC_TYPE_CONN_HEARTBEAT, TaskDto.SYNC_TYPE_LOG_COLLECTOR,TaskDto.SYNC_TYPE_MEM_CACHE})
 public class ObservableAspectTask extends AspectTask {
+	public static final String TAPCOMPLETABLEFUTUREEX_QUEUE_SIZE_PROP_KEY = String.join("_", TapCompletableFutureEx.TAG, "QUEUE", "SIZE");
+	public static final int TAPCOMPLETABLEFUTUREEX_QUEUE_SIZE_DEFAULT_VALUE = 500000;
+	public static final String TAPCOMPLETABLEFUTUREEX_JOIN_WATERMARK_PROP_KEY = String.join("_", TapCompletableFutureEx.TAG, "JOIN", "WATERMARK");
+	public static final int TAPCOMPLETABLEFUTUREEX_JOIN_WATERMARK_DEFAULT_VALUE = 500000;
 	private final ClassHandlers observerClassHandlers = new ClassHandlers();
 
 	private TaskSampleHandler taskSampleHandler;
 	private Map<String, TableSampleHandler> tableSampleHandlers;
 	private Map<String, DataNodeSampleHandler> dataNodeSampleHandlers;
 	private Map<String, ProcessorNodeSampleHandler> processorNodeSampleHandlers;
+	private static final String TAG = ObservableAspectTask.class.getSimpleName();
 
 	public ObservableAspectTask() {
 		// data node aspects
@@ -67,7 +78,7 @@ public ObservableAspectTask() {
 	CompletableFuture<Void> batchProcessFuture;
 	CompletableFuture<Void> streamReadFuture;
 	CompletableFuture<Void> streamProcessFuture;
-	CompletableFuture<Void> writeRecordFuture;
+	TapCompletableFutureEx writeRecordFuture;
 	/**
 	 * The task started
 	 */
@@ -83,15 +94,23 @@ protected void initCompletableFuture() {
 		batchProcessFuture = CompletableFuture.runAsync(()->{});
 		streamReadFuture = CompletableFuture.runAsync(()->{});
 		streamProcessFuture = CompletableFuture.runAsync(()->{});
-		writeRecordFuture = CompletableFuture.runAsync(()->{});
+		writeRecordFuture = TapCompletableFutureTaskEx.create(
+				CommonUtils.getPropertyInt(TAPCOMPLETABLEFUTUREEX_QUEUE_SIZE_PROP_KEY, TAPCOMPLETABLEFUTUREEX_QUEUE_SIZE_DEFAULT_VALUE),
+				CommonUtils.getPropertyInt(TAPCOMPLETABLEFUTUREEX_JOIN_WATERMARK_PROP_KEY, TAPCOMPLETABLEFUTUREEX_JOIN_WATERMARK_DEFAULT_VALUE),
+				task
+		).start();
 	}
 
 	protected void closeCompletableFuture() {
-		if (null != batchReadFuture) batchReadFuture.cancel(true);
-		if (null != batchProcessFuture) batchProcessFuture.cancel(true);
-		if (null != streamReadFuture) streamReadFuture.cancel(true);
-		if (null != streamProcessFuture) streamProcessFuture.cancel(true);
-		if (null != writeRecordFuture) writeRecordFuture.cancel(true);
+		if (null != batchReadFuture) batchReadFuture.cancel(false);
+		if (null != batchProcessFuture) batchProcessFuture.cancel(false);
+		if (null != streamReadFuture) streamReadFuture.cancel(false);
+		if (null != streamProcessFuture) streamProcessFuture.cancel(false);
+		try {
+			if (null != writeRecordFuture) writeRecordFuture.stop(15L, TimeUnit.SECONDS);
+		}catch (Exception e){
+			TapLogger.info(TAG, "Stop writeRecordFuture fail: {}", Log4jUtil.getStackString(e));
+		}
 	}
 
 	/**
@@ -118,8 +137,8 @@ public void onStop(TaskStopAspect stopAspect) {
 			}
 		}
 
-		taskSampleHandler.close();
 		closeCompletableFuture();
+		taskSampleHandler.close();
 	}
 
 	// data node related
@@ -307,6 +326,9 @@ public Void handleSourceDynamicTable(SourceDynamicTableAspect aspect) {
 	// target data node related
 
 	public Void handleCreateTableFunc(CreateTableFuncAspect aspect) {
+		if(aspect.isInit()){
+			return null;
+		}
 		String nodeId = aspect.getDataProcessorContext().getNode().getId();
 		switch (aspect.getState()) {
 			case CreateTableFuncAspect.STATE_START:
@@ -323,6 +345,9 @@ public Void handleCreateTableFunc(CreateTableFuncAspect aspect) {
 	}
 
 	public Void handleDropTableFunc(DropTableFuncAspect aspect) {
+		if(aspect.isInit()){
+			return null;
+		}
 		String nodeId = aspect.getDataProcessorContext().getNode().getId();
 		switch (aspect.getState()) {
 			case DropTableFuncAspect.STATE_START:
@@ -409,6 +434,7 @@ public Void handleCDCHeartbeatWriteAspect(CDCHeartbeatWriteAspect aspect) {
 	}
 
 	private PipelineDelayImpl pipelineDelay = (PipelineDelayImpl) InstanceFactory.instance(PipelineDelay.class);
+
 	public Void handleWriteRecordFunc(WriteRecordFuncAspect aspect) {
 		Node<?> node = aspect.getDataProcessorContext().getNode();
 		String nodeId = node.getId();
@@ -423,37 +449,41 @@ public Void handleWriteRecordFunc(WriteRecordFuncAspect aspect) {
 						}
 				);
 				aspect.consumer((events, result) -> {
-					writeRecordFuture.thenRunAsync(() -> {
-					if (null == events || events.size() == 0) {
-						return;
-					}
-
-					HandlerUtil.EventTypeRecorder inner = HandlerUtil.countTapEvent(events, recorder.getMemorySize());
-					Optional.ofNullable(dataNodeSampleHandlers.get(nodeId)).ifPresent(
-							handler -> {
-								handler.handleWriteRecordAccept(System.currentTimeMillis(), result, inner);
+					try {
+						writeRecordFuture.runAsync(() -> {
+							if (null == events || events.isEmpty()) {
+								return;
 							}
-					);
 
-					taskSampleHandler.handleWriteRecordAccept(result, events, inner);
-
-					Optional.ofNullable(tableSampleHandlers)
-							.flatMap(handlers -> {
-								// source >> target table name maybe change
-
-								String syncType = aspect.getDataProcessorContext().getTaskDto().getSyncType();
-								if (TaskDto.SYNC_TYPE_SYNC.equals(syncType) || TaskDto.SYNC_TYPE_CONN_HEARTBEAT.equals(syncType)) {
-									return handlers.values().stream().findFirst();
-								} else {
-									LinkedHashMap<String, String> tableNameRelation = ((DatabaseNode) node).getSyncObjects().get(0).getTableNameRelation();
-									String targetTableName = HashBiMap.create(tableNameRelation).inverse().get(table);
-									return Optional.ofNullable(handlers.get(targetTableName));
-								}
-							})
-							.ifPresent(handler -> handler.incrTableSnapshotInsertTotal(recorder.getInsertTotal()));
-
-					pipelineDelay.refreshDelay(task.getId().toHexString(), nodeId, inner.getProcessTimeTotal() / inner.getTotal(), inner.getNewestEventTimestamp());
-					});
+							HandlerUtil.EventTypeRecorder inner = HandlerUtil.countTapEvent(events, recorder.getMemorySize());
+							Optional.ofNullable(dataNodeSampleHandlers.get(nodeId)).ifPresent(
+									handler -> {
+										handler.handleWriteRecordAccept(System.currentTimeMillis(), result, inner);
+									}
+							);
+
+							taskSampleHandler.handleWriteRecordAccept(result, events, inner);
+
+							Optional.ofNullable(tableSampleHandlers)
+									.flatMap(handlers -> {
+										// source >> target table name maybe change
+
+										String syncType = aspect.getDataProcessorContext().getTaskDto().getSyncType();
+										if (TaskDto.SYNC_TYPE_SYNC.equals(syncType) || TaskDto.SYNC_TYPE_CONN_HEARTBEAT.equals(syncType)) {
+											return handlers.values().stream().findFirst();
+										} else {
+											LinkedHashMap<String, String> tableNameRelation = ((DatabaseNode) node).getSyncObjects().get(0).getTableNameRelation();
+											String targetTableName = HashBiMap.create(tableNameRelation).inverse().get(table);
+											return Optional.ofNullable(handlers.get(targetTableName));
+										}
+									})
+									.ifPresent(handler -> handler.incrTableSnapshotInsertTotal(recorder.getInsertTotal()));
+
+							pipelineDelay.refreshDelay(task.getId().toHexString(), nodeId, inner.getProcessTimeTotal() / inner.getTotal(), inner.getNewestEventTimestamp());
+						});
+					} catch (Exception e) {
+						TapLogger.info("Run async writeRecordFuture fail: {}", Log4jUtil.getStackString(e));
+					}
 				});
 				break;
 			case WriteRecordFuncAspect.STATE_END:
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureEx.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureEx.java
new file mode 100644
index 0000000000..b48c955714
--- /dev/null
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureEx.java
@@ -0,0 +1,179 @@
+package io.tapdata.observable.metric.util;
+
+import com.google.common.collect.Queues;
+import io.tapdata.entity.logger.TapLogger;
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author samuel
+ * @Description Asynchronous task executor based on CompletableFuture, submit Runnable through runAsync method.
+ * It will automatically execute asynchronously, and stop through stop method.
+ * This utility method does not guarantee execution order, but guarantees that all submitted tasks are executed before stopping.
+ * Usage example
+ * <pre>
+ *     TapCompletableFutureEx tapCompletableFutureEx = TapCompletableFutureEx.create(1000, 1000).start();
+ *     tapCompletableFutureEx.runAsync(() -> {
+ *     	// do something
+ *     });
+ * 	   tapCompletableFutureEx.runAsync(() -> {
+ *     	// do something
+ *     });
+ *     tapCompletableFutureEx.stop(1L, TimeUnit.SECONDS);
+ * </pre>
+ * @create 2024-06-28 12:01
+ **/
+public class TapCompletableFutureEx {
+	public static final String TAG = TapCompletableFutureEx.class.getSimpleName();
+	public static final String STOP_JOIN_CHECKER_THREAD_NAME = String.join("-", TAG, "stop", "join", "checker");
+	public static final String RUNNING_JOIN_CHECKER_THREAD_NAME = String.join("-", TAG, "running", "join", "checker");
+	public static final int DEFAULT_QUEUE_SIZE = 1000;
+	public static final int DEFAULT_JOIN_WATERMARK = 1000;
+
+	protected final AtomicBoolean running;
+	protected final int queueSize;
+	protected final LinkedBlockingQueue<CompletableFuture<Void>> completableFutureQueue;
+	protected final int joinWatermark;
+	protected ExecutorService joinThreadPool;
+	private final int[] joinLock = new int[0];
+	protected final AtomicBoolean firstTime = new AtomicBoolean(true);
+
+	protected TapCompletableFutureEx(int queueSize, int joinWatermark, String threadName) {
+		this.running = new AtomicBoolean(false);
+		this.queueSize = queueSize <= 0 ? DEFAULT_QUEUE_SIZE : queueSize;
+		this.joinWatermark = joinWatermark <= 0 ? DEFAULT_JOIN_WATERMARK : joinWatermark;
+		this.completableFutureQueue = new LinkedBlockingQueue<>(this.queueSize);
+		this.joinThreadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>(),
+				r -> new Thread(r, threadName));
+	}
+
+	public static TapCompletableFutureEx create(int queueSize, int joinWatermark) {
+		return new TapCompletableFutureEx(queueSize, joinWatermark, RUNNING_JOIN_CHECKER_THREAD_NAME);
+	}
+
+	public TapCompletableFutureEx start() {
+		this.running.compareAndSet(false, true);
+		this.joinThreadPool.execute(() -> {
+			while (running.get()) {
+				checkJoin();
+			}
+		});
+		return this;
+	}
+
+	public void runAsync(Runnable runnable) {
+		if (!running.get() && firstTime.compareAndSet(true, false)) {
+			throw new IllegalStateException("TapCompletableFutureEx not started");
+		}
+		enqueue(CompletableFuture.runAsync(runnable));
+	}
+
+	private void enqueue(CompletableFuture<Void> completableFuture) {
+		while (true) {
+			boolean isBreak = false;
+			try {
+				if (completableFutureQueue.offer(completableFuture, 10L, TimeUnit.MILLISECONDS)) {
+					isBreak = true;
+				}
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				isBreak = true;
+			}
+			if (isBreak) {
+				break;
+			}
+		}
+	}
+
+	protected void checkJoin() {
+		synchronized (joinLock) {
+			List<CompletableFuture<Void>> completableFutures = new ArrayList<>();
+			try {
+				Queues.drain(completableFutureQueue, completableFutures, joinWatermark, 1L, TimeUnit.SECONDS);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			if (CollectionUtils.isEmpty(completableFutures)) {
+				return;
+			}
+			CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join();
+		}
+	}
+
+	public void stop(long timeout, TimeUnit timeUnit) {
+		stopPrivate(timeout, timeUnit, STOP_JOIN_CHECKER_THREAD_NAME);
+	}
+
+	protected void stopPrivate(long timeout, TimeUnit timeUnit, String threadName) {
+		running.compareAndSet(true, false);
+		joinThreadPool.shutdownNow();
+		CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+			Thread.currentThread().setName(threadName);
+			stopCheckJoin();
+		});
+		waitDone(future, timeout, timeUnit);
+		Thread.currentThread().interrupt();
+	}
+
+	protected void stopCheckJoin() {
+		synchronized (joinLock) {
+			while (true) {
+				boolean isBreak = false;
+				List<CompletableFuture<Void>> completableFutureList = new ArrayList<>();
+				try {
+					Queues.drain(completableFutureQueue, completableFutureList, completableFutureQueue.size(), 1L, TimeUnit.SECONDS);
+				} catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					isBreak = true;
+				}
+				if (!completableFutureList.isEmpty()) {
+					CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[0])).join();
+				} else {
+					isBreak = true;
+				}
+				try {
+					TimeUnit.MILLISECONDS.sleep(100L);
+				} catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					isBreak = true;
+				}
+				if (isBreak) {
+					break;
+				}
+			}
+		}
+	}
+
+	private void waitDone(CompletableFuture<Void> future, long timeout, TimeUnit timeUnit) {
+		long timeoutMs = timeUnit.toMillis(timeout);
+		long s = System.currentTimeMillis();
+		while (true) {
+			long cost = System.currentTimeMillis() - s;
+			boolean futureDone = future.isDone();
+			boolean isBreak = cost > timeoutMs || futureDone;
+			try {
+				TimeUnit.MILLISECONDS.sleep(1L);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				isBreak = true;
+			}
+			if (isBreak) {
+				TapLogger.info(TAG, "Stop done, timeout: {} ms, cost time: {} ms, stop join checker isDone: {}", timeoutMs, cost, futureDone);
+				future.cancel(true);
+				break;
+			}
+		}
+	}
+
+	public int getQueueSize() {
+		return queueSize;
+	}
+
+	public int getJoinWatermark() {
+		return joinWatermark;
+	}
+}
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskEx.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskEx.java
new file mode 100644
index 0000000000..db9914d2f6
--- /dev/null
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskEx.java
@@ -0,0 +1,34 @@
+package io.tapdata.observable.metric.util;
+
+import com.tapdata.tm.commons.task.dto.TaskDto;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author samuel
+ * @Description
+ * @create 2024-06-28 12:12
+ **/
+public class TapCompletableFutureTaskEx extends TapCompletableFutureEx {
+	private final TaskDto taskDto;
+
+	protected TapCompletableFutureTaskEx(int queueSize, int clearWatermark, TaskDto taskDto) {
+		super(queueSize, clearWatermark, String.join("-", RUNNING_JOIN_CHECKER_THREAD_NAME,
+				(null != taskDto && null != taskDto.getName()) ? taskDto.getName() : "[no name]", (null != taskDto && null != taskDto.getId() ? taskDto.getId().toString() : "[no id]")));
+		this.taskDto = taskDto;
+	}
+
+	public static TapCompletableFutureTaskEx create(int queueSize, int joinWatermark, TaskDto taskDto) {
+		return new TapCompletableFutureTaskEx(queueSize, joinWatermark, taskDto);
+	}
+
+	@Override
+	public void stop(long timeout, TimeUnit timeUnit) {
+		stopPrivate(timeout, timeUnit, String.join("-", STOP_JOIN_CHECKER_THREAD_NAME,
+				(null != taskDto && null != taskDto.getName()) ? taskDto.getName() : "[no name]", (null != taskDto && null != taskDto.getId() ? taskDto.getId().toString() : "[no id]")));
+	}
+
+	public TaskDto getTaskDto() {
+		return taskDto;
+	}
+}
diff --git a/iengine/modules/proxy-client-module/src/main/java/io/tapdata/proxy/client/ProxySubscriptionManager.java b/iengine/modules/proxy-client-module/src/main/java/io/tapdata/proxy/client/ProxySubscriptionManager.java
index c89c8c40c7..cca1f2a1f5 100644
--- a/iengine/modules/proxy-client-module/src/main/java/io/tapdata/proxy/client/ProxySubscriptionManager.java
+++ b/iengine/modules/proxy-client-module/src/main/java/io/tapdata/proxy/client/ProxySubscriptionManager.java
@@ -61,7 +61,7 @@ public class ProxySubscriptionManager implements MemoryFetcher {
 	private final ConcurrentHashMap<String, TaskSubscribeInfo> taskIdTaskSubscribeInfoMap = new ConcurrentHashMap<>();
 //	private ScheduledFuture<?> workingFuture;
 //	private final AtomicBoolean needSync = new AtomicBoolean(false);
-	private IMClient imClient;
+	private volatile IMClient imClient;
 
 	private MaxFrequencyLimiter maxFrequencyLimiter;
 
@@ -101,7 +101,8 @@ public void startIMClient(List<String> baseURLs, String accessToken) {
 							baseUrl = baseUrl + "/";
 						newBaseUrls.add(baseUrl + "proxy");//?access_token=" + accessToken);
 					}
-					imClient = new IMClientBuilder()
+
+					IMClient imClientPrivate = new IMClientBuilder()
 							.withBaseUrl(newBaseUrls)
 							.withService("engine")
 							.withPrefix("e")
@@ -109,13 +110,14 @@ public void startIMClient(List<String> baseURLs, String accessToken) {
 							.withTerminal(1)
 							.withToken(accessToken)
 							.build();
-					imClient.start();
+					imClientPrivate.start();
 					EventManager eventManager = EventManager.getInstance();
-					eventManager.registerEventListener(imClient.getPrefix() + ".status", this::handleStatus);
+					eventManager.registerEventListener(imClientPrivate.getPrefix() + ".status", this::handleStatus);
 					//prefix + "." + data.getClass().getSimpleName() + "." + data.getContentType()
-					eventManager.registerEventListener(imClient.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + NewDataReceived.class.getSimpleName(), this::handleNewDataReceived);
-					eventManager.registerEventListener(imClient.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + CommandReceived.class.getSimpleName(), this::handleCommandReceived);
-					eventManager.registerEventListener(imClient.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + ServiceCallerReceived.class.getSimpleName(), this::handleServiceCallerReceived);
+					eventManager.registerEventListener(imClientPrivate.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + NewDataReceived.class.getSimpleName(), this::handleNewDataReceived);
+					eventManager.registerEventListener(imClientPrivate.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + CommandReceived.class.getSimpleName(), this::handleCommandReceived);
+					eventManager.registerEventListener(imClientPrivate.getPrefix() + "." + OutgoingData.class.getSimpleName() + "." + ServiceCallerReceived.class.getSimpleName(), this::handleServiceCallerReceived);
+					this.imClient = imClientPrivate;
 				}
 			}
 		}
diff --git a/iengine/validator/src/main/java/com/tapdata/validator/ConnectionValidator.java b/iengine/validator/src/main/java/com/tapdata/validator/ConnectionValidator.java
index 3b346c58ce..483330a592 100644
--- a/iengine/validator/src/main/java/com/tapdata/validator/ConnectionValidator.java
+++ b/iengine/validator/src/main/java/com/tapdata/validator/ConnectionValidator.java
@@ -118,8 +118,7 @@ public static boolean continueValidateConnection(BaseConnectionValidateResult pr
 	}
 
 	public static boolean validateHostPort(String databaseHost, Integer databasePort) {
-		try {
-			Socket s = new Socket();
+		try (Socket s = new Socket()) {
 			s.connect(new InetSocketAddress(databaseHost, databasePort), 10000);
 			s.close();
 			return true;
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/constant/InspectMethod.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/constant/InspectMethod.java
index 1f8605b7dd..10d980b8dd 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/constant/InspectMethod.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/constant/InspectMethod.java
@@ -1,9 +1,13 @@
 package com.tapdata.tm.inspect.constant;
 
 public enum InspectMethod {
+    ROW_COUNT("row_count"),
     FIELD("field"),
-    ROW_COUNT("row_count") ,
-    HASH("hash")  ;
+    JOINTFIELD("jointField"),
+    CDC_COUNT("cdcCount"),
+    JUNIT_TEST("junitTest"),
+    HASH("hash"),
+    ;
 
     private final String value;
 
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/dto/InspectDto.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/dto/InspectDto.java
index 780de78f5d..a8809e75a4 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/dto/InspectDto.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/dto/InspectDto.java
@@ -113,6 +113,8 @@ public class InspectDto extends SchedulableDto implements IDataPermissionDto {
 
     private String inspectResultId;     // 重新校验功能
     private List<String> taskIds;       // 重新校验要执行的任务id
+    private Boolean enableRecovery; // 启用数据恢复逻辑（不存库，传递给引擎启动）
+    private Boolean canRecovery; // 是否可恢复数据（不存库）
 
 		private List<AlarmSettingVO> alarmSettings;
 		private List<AlarmRuleVO> alarmRules;
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtils.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtils.java
new file mode 100644
index 0000000000..26cc9fa795
--- /dev/null
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtils.java
@@ -0,0 +1,38 @@
+package com.tapdata.tm.inspect.recovery;
+
+import com.tapdata.tm.inspect.constant.InspectMethod;
+import com.tapdata.tm.inspect.constant.InspectResultEnum;
+import com.tapdata.tm.inspect.constant.InspectStatusEnum;
+import com.tapdata.tm.inspect.dto.InspectDto;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 数据自动修复工具类
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/12 12:15 Create
+ */
+public class AutoRecoveryUtils {
+
+    public static List<String> checkCanRecovery(InspectDto inspectDto) {
+        List<String> errorCodes = new ArrayList<>();
+        if (!(
+            InspectMethod.FIELD.getValue().equals(inspectDto.getInspectMethod())
+                || InspectMethod.JOINTFIELD.getValue().equals(inspectDto.getInspectMethod())
+        )) {
+            errorCodes.add("Inspect.Recovery.NotFieldMethod");
+        }
+        if (!InspectStatusEnum.DONE.getValue().equals(inspectDto.getStatus())) {
+            errorCodes.add("Inspect.Recovery.StatusNotDone");
+        }
+        if (!InspectResultEnum.FAILED.getValue().equals(inspectDto.getResult())) {
+            errorCodes.add("Inspect.Recovery.ResultNotFound");
+        }
+        if (null == inspectDto.getFlowId() || inspectDto.getFlowId().isEmpty()) {
+            errorCodes.add("Inspect.Recovery.IsNotWithTask");
+        }
+        return errorCodes;
+    }
+}
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectDetailsService.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectDetailsService.java
index b1cbdde4a9..6bc7fcec21 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectDetailsService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectDetailsService.java
@@ -4,12 +4,31 @@
 import com.tapdata.tm.inspect.dto.InspectDetailsDto;
 import com.tapdata.tm.inspect.entity.InspectDetailsEntity;
 import com.tapdata.tm.inspect.repository.InspectDetailsRepository;
+import com.tapdata.tm.inspect.vo.FailTableAndRowsVo;
 import lombok.NonNull;
 import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.query.Criteria;
+
+import java.util.List;
 
 public abstract class InspectDetailsService extends BaseService<InspectDetailsDto, InspectDetailsEntity, ObjectId, InspectDetailsRepository> {
     public InspectDetailsService(@NonNull InspectDetailsRepository repository) {
         super(repository, InspectDetailsDto.class, InspectDetailsEntity.class);
     }
 
-}
\ No newline at end of file
+    public FailTableAndRowsVo totalFailTableAndRows(String inspectResultId) {
+        AggregationResults<FailTableAndRowsVo> results = repository.aggregate(Aggregation.newAggregation(
+            Aggregation.match(Criteria.where("inspectResultId").is(inspectResultId).and("source").exists(true))
+            , Aggregation.group("taskId").count().as("rows")
+            , Aggregation.group().count().as("tableTotals").sum("rows").as("rowsTotals")
+        ), FailTableAndRowsVo.class);
+        List<FailTableAndRowsVo> mappedResults = results.getMappedResults();
+        if (!mappedResults.isEmpty()) {
+            return mappedResults.get(0);
+        }
+        return null;
+    }
+
+}
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectService.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectService.java
index 2b7d84877d..f2d4accf68 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/service/InspectService.java
@@ -5,6 +5,7 @@
 import com.tapdata.tm.base.dto.Filter;
 import com.tapdata.tm.base.dto.Page;
 import com.tapdata.tm.base.dto.Where;
+import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.base.service.BaseService;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.config.security.UserDetail;
@@ -13,6 +14,8 @@
 import com.tapdata.tm.inspect.dto.InspectDto;
 import com.tapdata.tm.inspect.entity.InspectEntity;
 import com.tapdata.tm.inspect.repository.InspectRepository;
+import com.tapdata.tm.inspect.vo.InspectRecoveryStartVerifyVo;
+import com.tapdata.tm.webhook.enums.ConstVariable;
 import lombok.NonNull;
 import org.bson.types.ObjectId;
 import org.springframework.transaction.annotation.Transactional;
@@ -77,5 +80,12 @@ public long updateByWhere(Where where, InspectDto dto, UserDetail userDetail){
 
     public abstract List<InspectDto> findAllByIds(List<String> inspectIds);
 
-    public abstract void fieldHandler(List<Task> tasks,UserDetail userDetail);
+
+    public InspectRecoveryStartVerifyVo recoveryStartVerity(InspectDto inspectDto) {
+        throw new BizException(ConstVariable.TA_OSS_NON_SUPPORT_FUNCTION_EXCEPTION);
+    }
+
+    public InspectRecoveryStartVerifyVo recoveryStart(InspectDto inspectDto, UserDetail userDetail) {
+        throw new BizException(ConstVariable.TA_OSS_NON_SUPPORT_FUNCTION_EXCEPTION);
+    }
 }
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/FailTableAndRowsVo.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/FailTableAndRowsVo.java
new file mode 100644
index 0000000000..5f9d611262
--- /dev/null
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/FailTableAndRowsVo.java
@@ -0,0 +1,17 @@
+package com.tapdata.tm.inspect.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 错误表和行统计结果
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/12 14:58 Create
+ */
+@Data
+public class FailTableAndRowsVo implements Serializable {
+    private Integer tableTotals;
+    private Long rowsTotals;
+}
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/InspectRecoveryStartVerifyVo.java b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/InspectRecoveryStartVerifyVo.java
new file mode 100644
index 0000000000..9100aeaffb
--- /dev/null
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/inspect/vo/InspectRecoveryStartVerifyVo.java
@@ -0,0 +1,51 @@
+package com.tapdata.tm.inspect.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * 校验修复启动前校验
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/11 10:54 Create
+ */
+@Data
+public class InspectRecoveryStartVerifyVo implements Serializable {
+    /** 校验任务编号 */
+    private String inspectId;
+    /** 第一次校验结果编号 */
+    private String firstCheckId;
+    /** 引擎编号（跟随同步任务） */
+    private String agentId;
+    /** 任务编号 */
+    private String flowId;
+    /** 任务名称 */
+    private String flowName;
+    /** 任务类型 */
+    private String flowType;
+    /** 任务状态 */
+    private String flowStatus;
+    /** 任务里程碑阶段 */
+    private String flowMilestoneStage;
+    /** 任务延迟（毫秒） */
+    private Long flowDelay;
+    /** 差异阈值 */
+    private Integer diffLimit;
+    /** 差异总数 */
+    private Long diffTotals;
+    /** 源多数据 */
+    private Long sourceOnly;
+    /** 目标多数据 */
+    private Long targetOnly;
+    /** 恢复数据量 */
+    private Long recoveryDataTotals;
+    /** 恢复表数量 */
+    private Integer recoveryTableTotals;
+
+    /** 是否符合恢复条件 */
+    private Boolean canRecovery;
+    /** 校验项错误码 */
+    private List<String> errorCodes;
+}
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/task/param/SaveShareCacheParam.java b/manager/tm-api/src/main/java/com/tapdata/tm/task/param/SaveShareCacheParam.java
index 03d7a88fef..60a7d21cc0 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/task/param/SaveShareCacheParam.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/task/param/SaveShareCacheParam.java
@@ -14,6 +14,8 @@ public class SaveShareCacheParam {
     private Map dag;
 
     private List edges;
+    private Boolean shareCdcEnable;
+    private Boolean enforceShareCdc;
 
 
 }
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/task/vo/ShareCacheDetailVo.java b/manager/tm-api/src/main/java/com/tapdata/tm/task/vo/ShareCacheDetailVo.java
index 6d1959ebf4..9fb5bc4c8f 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/task/vo/ShareCacheDetailVo.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/task/vo/ShareCacheDetailVo.java
@@ -33,5 +33,7 @@ public class ShareCacheDetailVo extends BaseVo {
   private Integer maxMemory;
 
   private String externalStorageId;
+  private Boolean shareCdcEnable;
+  private Boolean enforceShareCdc;
 
 }
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/base/dto/BaseDto.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/base/dto/BaseDto.java
index 6136165f32..d2c4c87da3 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/base/dto/BaseDto.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/base/dto/BaseDto.java
@@ -25,6 +25,7 @@
 @EqualsAndHashCode
 @ToString
 public class BaseDto implements Serializable {
+    public static final String FIELD_USER_ID = "user_id";
 
 	@JsonSerialize( using = ObjectIdSerialize.class)
 	@JsonDeserialize( using = ObjectIdDeserialize.class)
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/DAG.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/DAG.java
index 3d8e0702a3..d326526155 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/DAG.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/DAG.java
@@ -186,27 +186,9 @@ private static DAG ignoreDisabledNode(Dag beforeDAG, boolean ignore) {
                     .forEach(resource -> tableNamesList.addAll(((DatabaseNode) resource).getTableNames()));
 
             ArrayList<String> objectNames = Lists.newArrayList(tableNamesList);
-            LinkedHashMap<String, String> tableNameRelation = Maps.newLinkedHashMap();
             LinkedList<Node> nodeLists = parseLinkedNode(beforeDAG);
 
-            if (CollectionUtils.isNotEmpty(nodeLists)) {
-                Map<String, TableRenameTableInfo> originalMap = new LinkedHashMap<>();
-                for (Node nodeList : nodeLists) {
-                    if (nodeList instanceof TableRenameProcessNode) {
-                        Map<String, TableRenameTableInfo> tableRenameTableInfoMap = ((TableRenameProcessNode) nodeList).originalMap();
-                        originalMap.putAll(tableRenameTableInfoMap);
-                    }
-                }
-                for (int i = 0; i < tableNamesList.size(); i++) {
-                    String tableName = tableNamesList.get(i);
-                    String currentTableName = tableName;
-                    if (originalMap.containsKey(tableName)) {
-                        currentTableName = originalMap.get(tableName).getCurrentTableName();
-                        objectNames.set(i, currentTableName);
-                    }
-                    tableNameRelation.put(tableName, currentTableName);
-                }
-            }
+            LinkedHashMap<String, String> tableNameRelation = generateTableNameRelation(nodeLists, objectNames);
 
             for (Node<?> node : nodes) {
                 if (node == null) {
@@ -241,6 +223,39 @@ private static DAG ignoreDisabledNode(Dag beforeDAG, boolean ignore) {
         return dag;
     }
 
+    public static LinkedHashMap<String, String> generateTableNameRelation(LinkedList<Node> nodeLists, ArrayList<String> tableNames) {
+        Map<String, String> renameNodeConvertTableName = new HashMap<>();
+        LinkedHashMap<String, String> tableNameRelation = new LinkedHashMap<>();
+        if (CollectionUtils.isNotEmpty(nodeLists)) {
+            for (Node nodeList : nodeLists) {
+                if (nodeList instanceof TableRenameProcessNode) {
+                    renameNodeConvertTableName = getConvertTableNameMap((TableRenameProcessNode) nodeList, tableNames);
+                }
+            }
+        }
+        for (int i = 0; i < tableNames.size(); i++) {
+            String originalName = tableNames.get(i);
+            if (renameNodeConvertTableName.containsKey(originalName)) {
+                String currentTableName = renameNodeConvertTableName.get(originalName);
+                tableNameRelation.put(originalName, currentTableName);
+                tableNames.set(i, currentTableName);
+            } else {
+                tableNameRelation.put(originalName, originalName);
+            }
+        }
+        return tableNameRelation;
+    }
+
+    public static Map<String, String> getConvertTableNameMap(TableRenameProcessNode renameProcessNode, List<String> tableNames) {
+        Map<String, String> tableRenameTableMap = new HashMap<>();
+        Map<String, TableRenameTableInfo> originalMap = renameProcessNode.originalMap();
+        for (int i = 0; i < tableNames.size(); i++) {
+            String originalName = tableNames.get(i);
+            String currentTableName = renameProcessNode.convertTableName(originalMap, originalName, false);
+            tableRenameTableMap.put(originalName, currentTableName);
+        }
+        return tableRenameTableMap;
+    }
     private static List<Node> filterDisabledNode(List<Node> nodes, Set<String> nodeIds) {
         return nodes.stream().filter(next -> {
             Map<String, Object> attrs = next.getAttrs();
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/bean/SourceDto.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/bean/SourceDto.java
index 969fcaad98..7c5c1de45f 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/bean/SourceDto.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/bean/SourceDto.java
@@ -1,6 +1,10 @@
 
 package com.tapdata.tm.commons.schema.bean;
 
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.tapdata.tm.commons.base.convert.ObjectIdDeserialize;
+import com.tapdata.tm.commons.base.convert.ObjectIdSerialize;
 import com.tapdata.tm.commons.schema.Schema;
 import com.tapdata.tm.commons.schema.Tag;
 import lombok.Data;
@@ -19,7 +23,8 @@ public class SourceDto {
 
     public static final String STATUS_INVALID = "invalid";
     public static final String STATUS_READY = "ready";
-
+    @JsonDeserialize(using = ObjectIdDeserialize.class)
+    @JsonSerialize(using = ObjectIdSerialize.class)
     @Field("id")
     @Indexed
     private ObjectId id;
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 717e02f31f..c3451f9818 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
@@ -35,7 +35,7 @@ public class ConnHeartbeatUtils {
      * @return can start heartbeat
      */
     public static boolean checkTask(@NonNull String taskType, @NonNull String taskSyncType) {
-        return StringUtils.containsAny(taskSyncType, TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC, TaskDto.SYNC_TYPE_LOG_COLLECTOR)  //syncType is migrate or sync
+        return StringUtils.containsAny(taskSyncType, TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC, TaskDto.SYNC_TYPE_LOG_COLLECTOR,TaskDto.SYNC_TYPE_MEM_CACHE)  //syncType is migrate or sync
                         && !ParentTaskDto.TYPE_INITIAL_SYNC.equals(taskType) //task type is not initial_sync
         ;
     }
diff --git a/manager/tm-oss/src/main/java/com/tapdata/tm/inspect/service/InspectServiceImpl.java b/manager/tm-oss/src/main/java/com/tapdata/tm/inspect/service/InspectServiceImpl.java
index 02ce5fef03..408e0bf40b 100644
--- a/manager/tm-oss/src/main/java/com/tapdata/tm/inspect/service/InspectServiceImpl.java
+++ b/manager/tm-oss/src/main/java/com/tapdata/tm/inspect/service/InspectServiceImpl.java
@@ -147,8 +147,4 @@ public List<InspectDto> findAllByIds(List<String> inspectIds) {
         throw new BizException(ConstVariable.TA_OSS_NON_SUPPORT_FUNCTION_EXCEPTION);
     }
 
-    @Override
-    public void fieldHandler(List<Task> tasks,UserDetail userDetail) {
-        throw new BizException(ConstVariable.TA_OSS_NON_SUPPORT_FUNCTION_EXCEPTION);
-    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/discovery/service/DiscoveryServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/discovery/service/DiscoveryServiceImpl.java
index ae809185cc..f7fa8538b5 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/discovery/service/DiscoveryServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/discovery/service/DiscoveryServiceImpl.java
@@ -2,6 +2,7 @@
 import com.google.common.collect.Lists;
 import com.tapdata.tm.apiServer.dto.ApiServerDto;
 import com.tapdata.tm.apiServer.service.ApiServerService;
+import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.cluster.dto.ClusterStateDto;
 import com.tapdata.tm.cluster.service.ClusterStateService;
 import com.tapdata.tm.commons.dag.Edge;
@@ -25,6 +26,8 @@
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.discovery.bean.*;
 import com.tapdata.tm.ds.service.impl.DataSourceService;
+import com.tapdata.tm.livedataplatform.dto.LiveDataPlatformDto;
+import com.tapdata.tm.livedataplatform.service.LiveDataPlatformService;
 import com.tapdata.tm.metadatadefinition.dto.MetadataDefinitionDto;
 import com.tapdata.tm.metadatadefinition.service.MetadataDefinitionService;
 import com.tapdata.tm.metadatainstance.repository.MetadataInstancesRepository;
@@ -50,6 +53,7 @@
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.aggregation.*;
 import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Meta;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.stereotype.Service;
@@ -62,6 +66,8 @@
 @Setter(onMethod_ = {@Autowired})
 public class DiscoveryServiceImpl implements DiscoveryService {
 
+    public static final String SOURCE_ID = "source._id";
+    public static final String COUNT = "count";
     private MetadataInstancesService metadataInstancesService;
 
     private MetadataInstancesRepository metaDataRepository;
@@ -82,6 +88,8 @@ public class DiscoveryServiceImpl implements DiscoveryService {
 
     private LdpService ldpService;
 
+    private LiveDataPlatformService liveDataPlatformService;
+
     /**
      * 查询对象概览列表
      *
@@ -1012,7 +1020,6 @@ public Page<DataDirectoryDto> findDataDirectory(DirectoryQueryParam param, UserD
                 .and("is_deleted").ne(true);
         if (StringUtils.isNotBlank(param.getObjType())) {
             metadataCriteria.and("meta_type").is(param.getObjType());
-            //taskCriteria.and("syncType").is(param.getObjType());
             apiCriteria.and("apiType").is(param.getObjType());
         } else {
             metadataCriteria.and("meta_type").is("table");
@@ -1043,6 +1050,7 @@ public Page<DataDirectoryDto> findDataDirectory(DirectoryQueryParam param, UserD
         MetadataDefinitionDto definitionDto = null;
         if (StringUtils.isNotBlank(param.getTagId())) {
             definitionDto = metadataDefinitionService.findById(MongoUtils.toObjectId(param.getTagId()));
+            addMetadataCriteriaMDMConnId(user, metadataCriteria, definitionDto);
             if (definitionDto != null) {
                 List<String> itemTypes = definitionDto.getItemType();
 
@@ -1064,7 +1072,9 @@ public Page<DataDirectoryDto> findDataDirectory(DirectoryQueryParam param, UserD
                         taskCriteria.and("syncType").is(param.getObjType());
                     }
                     List<String> tagIds = andChild.stream().map(t->t.getId().toHexString()).collect(Collectors.toList());
-                    metadataCriteria.and("listtags.id").in(tagIds);
+                    if (StringUtils.isBlank(param.getQueryKey())) {
+                        addMetadataCriteriaListTags(definitionDto, tagIds, metadataCriteria);
+                    }
                     taskCriteria.and("listtags.id").in(tagIds);
                     apiCriteria.and("listtags.id").in(tagIds);
                 } else {
@@ -1076,7 +1086,7 @@ public Page<DataDirectoryDto> findDataDirectory(DirectoryQueryParam param, UserD
                                 apiCriteria.and("_id").is("1231231231");
                                 List<String> linkIds = andChild.stream().map(MetadataDefinitionDto::getLinkId).filter(Objects::nonNull).collect(Collectors.toList());
                                 if (CollectionUtils.isNotEmpty(linkIds)) {
-                                    metadataCriteria.and("source._id").in(linkIds);
+                                    metadataCriteria.and(SOURCE_ID).in(linkIds);
                                 } else {
                                     return page;
                                 }
@@ -1268,136 +1278,57 @@ public Page<DataDirectoryDto> findDataDirectory(DirectoryQueryParam param, UserD
         return page;
     }
 
-    //@Override
-    public Page<DataDirectoryDto> findDataDirectory1(DirectoryQueryParam param, UserDetail user) {
-        if (param.getPage() == null) {
-            param.setPage(1);
+    protected void addMetadataCriteriaListTags(MetadataDefinitionDto definitionDto, List<String> tagIds, Criteria metadataCriteria) {
+        if (isMDMRoot(definitionDto)) {
+            List<Criteria> metaOr = new ArrayList<>();
+            metaOr.add(Criteria.where("listtags.id").exists(false));
+            if (null != tagIds) {
+                metaOr.add(Criteria.where("listtags.id").in(tagIds));
+            }
+            metadataCriteria.orOperator(metaOr);
+        } else {
+            if (null != tagIds) {
+                metadataCriteria.and("listtags.id").in(tagIds);
+            }
         }
+    }
 
-        if (param.getPageSize() == null) {
-            param.setPageSize(20);
+    protected void addMetadataCriteriaMDMConnId(UserDetail user, Criteria metadataCriteria, MetadataDefinitionDto metadataDefinitionDto) {
+        if (null == metadataDefinitionDto) {
+            return;
         }
-
-        Page<DataDirectoryDto> page = new Page<>();
-        page.setItems(Lists.newArrayList());
-        page.setTotal(0);
-
-        Criteria taskCriteria = Criteria.where("is_deleted").ne(true);
-        Criteria apiCriteria = Criteria.where("status").is("active");
-
-        Criteria metadataCriteria = Criteria.where("sourceType").is(SourceTypeEnum.SOURCE.name())
-                .and("taskId").exists(false)
-                .and("is_deleted").ne(true);
-        if (StringUtils.isNotBlank(param.getObjType())) {
-            metadataCriteria.and("meta_type").is(param.getObjType());
-            taskCriteria.and("syncType").is(param.getObjType());
-            apiCriteria.and("apiType").is(param.getObjType());
-        } else {
-            metadataCriteria.and("meta_type").is("table");
+        List<String> itemType = metadataDefinitionDto.getItemType();
+        if (null == itemType || (!itemType.contains(MetadataDefinitionDto.LDP_ITEM_FDM) && !itemType.contains(MetadataDefinitionDto.LDP_ITEM_MDM))) {
+            return;
         }
-
-        if (StringUtils.isNotBlank(param.getQueryKey())) {
-            metadataCriteria.orOperator(
-                    Criteria.where("original_name").regex(param.getQueryKey()),
-                    Criteria.where("name").regex(param.getQueryKey()),
-                    Criteria.where("comment").regex(param.getQueryKey()),
-                    Criteria.where("source.name").regex(param.getQueryKey()),
-                    Criteria.where("alias_name").regex(param.getQueryKey()));
-
-            taskCriteria.orOperator(
-                    Criteria.where("name").regex(param.getQueryKey()),
-                    Criteria.where("desc").regex(param.getQueryKey()));
-
-            apiCriteria.orOperator(
-                    Criteria.where("name").regex(param.getQueryKey()),
-                    Criteria.where("tableName").regex(param.getQueryKey()));
+        LiveDataPlatformDto liveDataPlatformDto = liveDataPlatformService.findOne(Query.query(Criteria.where("user_id").is(user.getUserId())));
+        if (null == liveDataPlatformDto) {
+			throw new BizException("Ldp.not.exists", "Live data platform not exists, user: " + user.getUserId());
         }
-
-
-        if (StringUtils.isNotBlank(param.getTagId())) {
-            MetadataDefinitionDto definitionDto = metadataDefinitionService.findById(MongoUtils.toObjectId(param.getTagId()));
-            if (definitionDto != null) {
-                List<String> itemTypes = definitionDto.getItemType();
-                boolean isDefault = itemTypes.contains("default");
-                List<MetadataDefinitionDto> andChild = metadataDefinitionService.findAndChild(Lists.newArrayList(MongoUtils.toObjectId(param.getTagId())));
-                if (!isDefault) {
-
-                    if (StringUtils.isBlank(param.getObjType())) {
-                        taskCriteria.and("syncType").in(TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC);
-                    }
-                    List<String> tagIds = andChild.stream().map(t->t.getId().toHexString()).collect(Collectors.toList());
-                    metadataCriteria.and("listtags.id").in(tagIds);
-                    taskCriteria.and("listtags.id").in(tagIds);
-                    apiCriteria.and("listtags.id").in(tagIds);
-                } else {
-                    if (!definitionDto.getValue().equals("Root")) {
-                        DataObjCategoryEnum defaultObjEnum = getDefaultObjEnum(null, definitionDto.getId());
-                        switch (defaultObjEnum) {
-                            case storage:
-                                taskCriteria.and("_id").is("1231231231");
-                                apiCriteria.and("_id").is("1231231231");
-                                List<String> linkIds = andChild.stream().map(MetadataDefinitionDto::getLinkId).filter(Objects::nonNull).collect(Collectors.toList());
-                                if (CollectionUtils.isNotEmpty(linkIds)) {
-                                    metadataCriteria.and("source._id").in(linkIds);
-                                } else {
-                                    return page;
-                                }
-                                break;
-                            case job:
-                                metadataCriteria.and("_id").is("1231231231");
-                                apiCriteria.and("_id").is("1231231231");
-                                if ("sync".equals(definitionDto.getValue())) {
-                                    taskCriteria.and("syncType").is(TaskDto.SYNC_TYPE_SYNC);
-                                } else if ("migrate".equals(definitionDto.getValue())) {
-                                    taskCriteria.and("syncType").is(TaskDto.SYNC_TYPE_MIGRATE);
-                                } else {
-                                    taskCriteria.and("syncType").in(TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC);
-                                }
-                                break;
-                            case api:
-                                metadataCriteria.and("_id").is("1231231231");
-                                taskCriteria.and("_id").is("1231231231");
-                                break;
-                        }
-                    }
-                }
+        List<String> sourceIds = new ArrayList<>();
+        if (itemType.contains(MetadataDefinitionDto.LDP_ITEM_FDM)) {
+            String fdmStorageConnectionId = liveDataPlatformDto.getFdmStorageConnectionId();
+            if(StringUtils.isNotBlank(fdmStorageConnectionId)){
+                sourceIds.add(fdmStorageConnectionId);
             }
         }
-
-        UnionWithOperation taskUnion = UnionWithOperation.unionWith("TaskCollectionObj")
-                .pipeline(
-                        Aggregation.project("createTime", "_id", "listtags", "syncType", "name", "agentId", "is_deleted"),
-                        Aggregation.match(taskCriteria)
-                );
-
-        UnionWithOperation apiUnion = UnionWithOperation.unionWith("Modules")
-                .pipeline(
-                        Aggregation.project("createTime", "_id", "listtags", "name", "apiType", "tableName", "status"),
-                        Aggregation.match(apiCriteria)
-                );
-        MatchOperation match = Aggregation.match(metadataCriteria);
-        ProjectionOperation project = Aggregation.project("createTime", "_id", "listtags", "meta_type", "original_name"
-                , "source", "syncType", "name", "agentId", "apiType", "tableName", "comment", "desc");
-        LimitOperation limitOperation = Aggregation.limit(param.getPageSize());
-        SkipOperation skipOperation = Aggregation.skip((long) (param.getPage() - 1) * param.getPageSize());
-        SortOperation sortOperation = Aggregation.sort(Sort.Direction.DESC, "createTime");
-
-        long count1 = metadataInstancesService.count(new Query(metadataCriteria), user);
-        long count2 = taskRepository.count(new Query(taskCriteria), user);
-        long count3 = modulesService.count(new Query(apiCriteria), user);
-        long total = count1 + count2 + count3;
-        Aggregation aggregation = Aggregation.newAggregation(match, taskUnion, apiUnion, project, sortOperation, skipOperation, limitOperation);
-        AggregationResults<UnionQueryResult> results = metaDataRepository.getMongoOperations().aggregate(aggregation, "MetadataInstances", UnionQueryResult.class);
-        List<UnionQueryResult> unionQueryResults = results.getMappedResults();
-        List<DataDirectoryDto> items = unionQueryResults.stream().map(this::convertToDataDirectory).collect(Collectors.toList());
-
-        page.setItems(items);
-        page.setTotal(total);
-        return page;
+        if (itemType.contains(MetadataDefinitionDto.LDP_ITEM_MDM)) {
+            String mdmStorageConnectionId = liveDataPlatformDto.getMdmStorageConnectionId();
+            if(StringUtils.isNotBlank(mdmStorageConnectionId)){
+                sourceIds.add(mdmStorageConnectionId);
+            }
+        }
+        if (CollectionUtils.isNotEmpty(sourceIds)) {
+            metadataCriteria.and(SOURCE_ID).in(sourceIds);
+        }
     }
 
-
-
+    protected boolean isMDMRoot(MetadataDefinitionDto definitionDto) {
+        if (null == definitionDto) {
+            return false;
+        }
+        return StringUtils.isBlank(definitionDto.getParent_id()) && definitionDto.getItemType().contains(MetadataDefinitionDto.LDP_ITEM_MDM);
+    }
 
     private List<String> objCategoryFilterList() {
         return Arrays.stream(DataObjCategoryEnum.values()).map(Enum::name).collect(Collectors.toList());
@@ -1509,9 +1440,9 @@ public void addObjCount(List<MetadataDefinitionDto> tagDtos, UserDetail user) {
                 .and("taskId").exists(false)
                 .and("is_deleted").ne(true)
                 .and("meta_type").is("table")
-                .and("source._id").ne(null);
+                .and(SOURCE_ID).ne(null);
         MatchOperation match = Aggregation.match(criteria1);
-        GroupOperation g = Aggregation.group("source._id").count().as("count");
+        GroupOperation g = Aggregation.group(SOURCE_ID).count().as(COUNT);
 
 
         Aggregation aggregation = Aggregation.newAggregation(match, g);
@@ -1534,7 +1465,7 @@ public void addObjCount(List<MetadataDefinitionDto> tagDtos, UserDetail user) {
             criteriaSyncTask.and("user_id").is(user.getUserId());
         }
         MatchOperation matchTask = Aggregation.match(criteriaSyncTask);
-        GroupOperation gTask = Aggregation.group("user_id").count().as("count");
+        GroupOperation gTask = Aggregation.group("user_id").count().as(COUNT);
 
 
         Aggregation aggregationTask = Aggregation.newAggregation(matchTask, gTask);
@@ -1556,7 +1487,7 @@ public void addObjCount(List<MetadataDefinitionDto> tagDtos, UserDetail user) {
             criteriaMigrateTask.and("user_id").is(user.getUserId());
         }
         MatchOperation matchMigrateTask = Aggregation.match(criteriaMigrateTask);
-        GroupOperation gMigrateTask = Aggregation.group("user_id").count().as("count");
+        GroupOperation gMigrateTask = Aggregation.group("user_id").count().as(COUNT);
 
 
         Aggregation aggregationMigrateTask = Aggregation.newAggregation(matchMigrateTask, gMigrateTask);
diff --git a/manager/tm/src/main/java/com/tapdata/tm/inspect/controller/InspectController.java b/manager/tm/src/main/java/com/tapdata/tm/inspect/controller/InspectController.java
index aab0388c2e..fdcafab9f8 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/inspect/controller/InspectController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/inspect/controller/InspectController.java
@@ -17,6 +17,7 @@
 import com.tapdata.tm.inspect.dto.InspectDto;
 import com.tapdata.tm.inspect.service.InspectService;
 import com.tapdata.tm.inspect.service.InspectTaskService;
+import com.tapdata.tm.inspect.vo.InspectRecoveryStartVerifyVo;
 import com.tapdata.tm.inspect.vo.InspectTaskVo;
 import com.tapdata.tm.permissions.DataPermissionHelper;
 import com.tapdata.tm.permissions.constants.DataPermissionActionEnums;
@@ -37,15 +38,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.web.bind.annotation.DeleteMapping;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PatchMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
 
 import javax.servlet.http.HttpServletRequest;
 import java.util.Date;
@@ -136,11 +129,6 @@ public ResponseMessage<InspectDto> save(@RequestBody InspectDto inspect) {
             throw new BizException("Inspect.task.null");
         }
 
-        if(StringUtils.isNotEmpty(inspect.getInspectMethod()) &&
-                (inspect.getInspectMethod().equals("field") || inspect.getInspectMethod().equals("jointField"))){
-            inspectService.fieldHandler(task,getLoginUser());
-        }
-
         if (null == platformInfo || StringUtils.isEmpty(platformInfo.getAgentType())) {
             throw new BizException("Inspect.agentTag.null");
         }
@@ -387,4 +375,45 @@ public ResponseMessage<InspectDto> upsertByWhere(@RequestParam("where") String w
         return success(inspect);
     }
 
+
+    @Operation(summary = "启动差异修复")
+    @PutMapping("/{id}/recovery/start")
+    public ResponseMessage<InspectRecoveryStartVerifyVo> recoveryStart(HttpServletRequest request, @PathVariable String id) {
+        InspectDto inspectDto = dataPermissionCheckOfId(
+            request,
+            getLoginUser(),
+            MongoUtils.toObjectId(id),
+            DataPermissionActionEnums.Start,
+            Lists.newArrayList(DataPermissionActionEnums.Start),
+            () -> inspectService.findOne(Query.query(Criteria.where("_id").is(MongoUtils.toObjectId(id))))
+        );
+
+        if (null == inspectDto) {
+            throw new BizException("Inspect.NotFound");
+        }
+
+        UserDetail userDetail = getLoginUser();
+        InspectRecoveryStartVerifyVo startVerifyVo = inspectService.recoveryStart(inspectDto, userDetail);
+        return success(startVerifyVo);
+    }
+
+    @Operation(summary = "差异修复-验证信息")
+    @GetMapping("/{id}/recovery/start-verify")
+    public ResponseMessage<InspectRecoveryStartVerifyVo> recoveryStartVerify(HttpServletRequest request, @PathVariable String id) {
+        InspectDto inspectDto = dataPermissionCheckOfId(
+            request,
+            getLoginUser(),
+            MongoUtils.toObjectId(id),
+            DataPermissionActionEnums.Start,
+            Lists.newArrayList(DataPermissionActionEnums.Start),
+            () -> inspectService.findOne(Query.query(Criteria.where("_id").is(MongoUtils.toObjectId(id))))
+        );
+
+        if (null == inspectDto) {
+            throw new BizException("Inspect.NotFound");
+        }
+
+        InspectRecoveryStartVerifyVo startVerifyVo = inspectService.recoveryStartVerity(inspectDto);
+        return success(startVerifyVo);
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/livedataplatform/service/LiveDataPlatformService.java b/manager/tm/src/main/java/com/tapdata/tm/livedataplatform/service/LiveDataPlatformService.java
index d734a665ef..57c7bcfbb4 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/livedataplatform/service/LiveDataPlatformService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/livedataplatform/service/LiveDataPlatformService.java
@@ -11,6 +11,7 @@
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.stereotype.Service;
 
@@ -36,7 +37,8 @@ protected void beforeSave(LiveDataPlatformDto liveDataPlatformDto, UserDetail us
     public LiveDataPlatformDto save(LiveDataPlatformDto dto, UserDetail user) {
         Query query = new Query();
         query.fields().include("_id");
-        LiveDataPlatformDto old = findOne(new Query(), user);
+        query.addCriteria(Criteria.where("user_id").is(user.getUserId()));
+        LiveDataPlatformDto old = findOne(query, user);
         if (old != null) {
             dto.setId(old.getId());
         }
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 b2d14c958c..42eee0df6d 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
@@ -2021,6 +2021,8 @@ public ShareCacheDetailVo findShareCacheById(String id) {
         shareCacheDetailVo.setMaxMemory(targetNode.getMaxMemory());
         shareCacheDetailVo.setTtl(TimeUtil.parseSecondsToDay(targetNode.getTtl()));
         shareCacheDetailVo.setExternalStorageId(targetNode.getExternalStorageId());
+        shareCacheDetailVo.setShareCdcEnable(taskDto.getShareCdcEnable());
+        shareCacheDetailVo.setEnforceShareCdc(taskDto.getEnforceShareCdc());
 
         return shareCacheDetailVo;
     }
@@ -2032,7 +2034,8 @@ protected TaskDto parseCacheToTaskDto(SaveShareCacheParam saveShareCacheParam, T
         taskDto.setShareCache(true);
         taskDto.setLastUpdAt(new Date());
         taskDto.setName(saveShareCacheParam.getName());
-        taskDto.setShareCdcEnable(false);
+        taskDto.setShareCdcEnable(saveShareCacheParam.getShareCdcEnable());
+        taskDto.setEnforceShareCdc(saveShareCacheParam.getEnforceShareCdc());
 
         DAG dag = taskDto.getDag();
         String sourceId;
diff --git a/manager/tm/src/main/resources/messages.properties b/manager/tm/src/main/resources/messages.properties
index 7a94383ca7..fab104a823 100644
--- a/manager/tm/src/main/resources/messages.properties
+++ b/manager/tm/src/main/resources/messages.properties
@@ -82,6 +82,14 @@ Inspect.Name.Exist=inspect name already exist
 Inspect.agentTag.null=inspect agent must not be null
 Inspect.task.null=inspect task must not be null
 Inspect.ProcessId.NotFound=inspect processid notfound
+Inspect.Recovery.NotFieldMethod = Not field inspection task
+Inspect.Recovery.StatusNotDone = Inspection task is not in a 'done' status
+Inspect.Recovery.ResultNotFound = Inspection task has no discrepancy results
+Inspect.Recovery.IsNotWithTask = Inspection is not created based on a synchronization task
+Inspect.Recovery.TaskNotFound = Synchronization task does not exist
+Inspect.Recovery.TaskTypeIsNotCDC = Synchronization task does not include incremental phase
+Inspect.Recovery.TaskIsNotRunning = Synchronization task is not in a 'running' status
+Inspect.Recovery.TaskSyncStatusIsNotCDC = Synchronization task is not in the incremental phase
 
 Function.Name.Exist=function name already exist
 Illegal.license=Illegal license
diff --git a/manager/tm/src/main/resources/messages_en_US.properties b/manager/tm/src/main/resources/messages_en_US.properties
index cfb9cb4aa8..61654f3688 100644
--- a/manager/tm/src/main/resources/messages_en_US.properties
+++ b/manager/tm/src/main/resources/messages_en_US.properties
@@ -15,6 +15,14 @@ Inspect.Name.Exist=name already exists
 Inspect.ProcessId.NotFound=inspect processid notfound
 Inspect.agentTag.null=inspect agent must not be null
 Inspect.task.null=inspect task must not be null
+Inspect.Recovery.NotFieldMethod = Not field inspection task
+Inspect.Recovery.StatusNotDone = Inspection task is not in a 'done' status
+Inspect.Recovery.ResultNotFound = Inspection task has no discrepancy results
+Inspect.Recovery.IsNotWithTask = Inspection is not created based on a synchronization task
+Inspect.Recovery.TaskNotFound = Synchronization task does not exist
+Inspect.Recovery.TaskTypeIsNotCDC = Synchronization task does not include incremental phase
+Inspect.Recovery.TaskIsNotRunning = Synchronization task is not in a 'running' status
+Inspect.Recovery.TaskSyncStatusIsNotCDC = Synchronization task is not in the incremental phase
 
 AccessCode.Is.Null=AccessCode is null
 AccessCode.No.User=can not find user by this accessCode
diff --git a/manager/tm/src/main/resources/messages_zh_CN.properties b/manager/tm/src/main/resources/messages_zh_CN.properties
index 3f2c4f3bf6..e53c16330f 100644
--- a/manager/tm/src/main/resources/messages_zh_CN.properties
+++ b/manager/tm/src/main/resources/messages_zh_CN.properties
@@ -26,6 +26,14 @@ Inspect.Name.Exist=\u540D\u79F0\u5DF2\u5B58\u5728\uFF0C\u8BF7\u4FEE\u6539
 Inspect.ProcessId.NotFound= \u627E\u4E0D\u5230\u5BF9\u5E94\u7684agent
 Inspect.agentTag.null=tag\u4E0D\u80FD\u4E3A\u7A7A
 Inspect.task.null=task\u4E0D\u80FD\u4E3A\u7A7A
+Inspect.Recovery.NotFieldMethod=\u975e\u5b57\u6bb5\u6821\u9a8c\u4efb\u52a1
+Inspect.Recovery.StatusNotDone=\u6821\u9a8c\u4efb\u52a1\u4e0d\u662f\u5df2\u5b8c\u6210\u72b6\u6001
+Inspect.Recovery.ResultNotFound=\u6821\u9a8c\u4efb\u52a1\u6ca1\u6709\u5dee\u5f02\u7ed3\u679c
+Inspect.Recovery.IsNotWithTask=\u6821\u9a8c\u4e0d\u662f\u57fa\u4e8e\u540c\u6b65\u4efb\u52a1\u521b\u5efa
+Inspect.Recovery.TaskNotFound=\u540c\u6b65\u4efb\u52a1\u4e0d\u5b58\u5728
+Inspect.Recovery.TaskTypeIsNotCDC=\u540c\u6b65\u4efb\u52a1\u4e0d\u5305\u542b\u589e\u91cf
+Inspect.Recovery.TaskIsNotRunning=\u540c\u6b65\u4efb\u52a1\u975e\u8fd0\u884c\u72b6\u6001
+Inspect.Recovery.TaskSyncStatusIsNotCDC=\u540c\u6b65\u4efb\u52a1\u4e0d\u5728\u589e\u91cf\u9636\u6bb5
 
 
 
diff --git a/manager/tm/src/main/resources/messages_zh_TW.properties b/manager/tm/src/main/resources/messages_zh_TW.properties
index 0b14e3a24f..bbcc9d7b4d 100644
--- a/manager/tm/src/main/resources/messages_zh_TW.properties
+++ b/manager/tm/src/main/resources/messages_zh_TW.properties
@@ -6,6 +6,15 @@ InvalidPaidPlan=\u7121\u6548\u7684\u4ED8\u8CBB\u65B9\u6848\uFF0C\u8ACB\u6AA2\u67
 Task.CronError = \u8ABF\u5EA6\u904B\u7B97\u5F0F\u932F\u8AA4
 Role.Unable.Delete=\u7121\u6CD5\u522A\u9664,\u7576\u524D\u89D2\u8272\u5B58\u5728\u7D81\u5B9A\u7528\u6236
 
+Inspect.Recovery.NotFieldMethod=\u975e\u5b57\u6bb5\u6821\u9a57\u4efb\u52d9
+Inspect.Recovery.StatusNotDone=\u6821\u9a57\u4efb\u52d9\u4e0d\u662f\u5df2\u5b8c\u6210\u72c0\u614b
+Inspect.Recovery.ResultNotFound=\u6821\u9a57\u4efb\u52d9\u6c92\u6709\u5dee\u7570\u7d50\u679c
+Inspect.Recovery.IsNotWithTask=\u6821\u9a57\u4e0d\u662f\u57fa\u65bc\u540c\u6b65\u4efb\u52d9\u5275\u5efa
+Inspect.Recovery.TaskNotFound=\u540c\u6b65\u4efb\u52d9\u4e0d\u5b58\u5728
+Inspect.Recovery.TaskTypeIsNotCDC=\u540c\u6b65\u4efb\u52d9\u4e0d\u5305\u542b\u589e\u91cf
+Inspect.Recovery.TaskIsNotRunning=\u540c\u6b65\u4efb\u52d9\u975e\u904b\u884c\u72c0\u614b
+Inspect.Recovery.TaskSyncStatusIsNotCDC=\u540c\u6b65\u4efb\u52d9\u4e0d\u5728\u589e\u91cf\u968e\u6bb5
+
 # external storage
 External.Storage.Name.Blank=\u540D\u7A31\u4E0D\u80FD\u70BA\u7A7A
 External.Storage.Name.Exists=\u540D\u7A31 {0} \u5DF2\u7D93\u5B58\u5728
diff --git a/pom.xml b/pom.xml
index a1ccb09d67..ffa977a373 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,6 +20,7 @@
             manager/**/dto/*,
             manager/**/entity/*,
             manager/**/vo/*,
+            manager/**/param/*,
             manager/tm/**/controller/*,
             manager/**/constant/*,
             manager/tm/**/ex/*,
@@ -33,16 +34,21 @@
             manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/*,
             **/*Entity.java,
             **/*VO.java,
+            **/src/main/java/io/tapdata/**/exception/**,
+            iengine/api/src/main/java/io/tapdata/aspect/CreateTableFuncAspect.java,
+            iengine/api/src/main/java/io/tapdata/aspect/DropTableFuncAspect.java
         </sonar.coverage.exclusions>
         <sonar.exclusions>
             manager/**/dto/*,
             manager/**/entity/*,
             manager/**/vo/*,
+            manager/**/param/*,
             manager/tm/**/controller/*,
             manager/tm/**/constant/*,
             manager/tm/**/ex/*,
             **/*Dto.java,
             **/*Enum.java,
+            **/src/main/java/io/tapdata/**/exception/**,
             iengine/iengine-common/src/main/java/com/tapdata/entity/**,
             tapdata-cli/**,
             manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/**,


Test patch:
diff --git a/iengine/api/src/test/java/io/tapdata/inspect/AutoRecoveryTest.java b/iengine/api/src/test/java/io/tapdata/inspect/AutoRecoveryTest.java
new file mode 100644
index 0000000000..cd42476737
--- /dev/null
+++ b/iengine/api/src/test/java/io/tapdata/inspect/AutoRecoveryTest.java
@@ -0,0 +1,162 @@
+package io.tapdata.inspect;
+
+import com.tapdata.entity.TapdataRecoveryEvent;
+import io.tapdata.inspect.exception.DuplicateAutoRecoveryException;
+import io.tapdata.inspect.exception.DuplicateClientAutoRecoveryException;
+import io.tapdata.inspect.exception.NotfoundAutoRecoveryException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 自动恢复数据测试
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/14 10:32 Create
+ */
+class AutoRecoveryTest {
+    private static final String taskId = "test-task-id";
+    private static final String inspectTaskId = "test-inspect-task-id";
+    private static final long interval = 50;
+
+    @Test
+    void testDuplicateServer() {
+        Assertions.assertNotNull(Assertions.assertThrows(DuplicateAutoRecoveryException.class, () -> {
+            try (AutoRecovery ignore1 = AutoRecovery.init(taskId);
+                 AutoRecovery ignore2 = AutoRecovery.init(taskId)) {
+                // do nothing
+            }
+        }), "Duplicate auto-recovery detected");
+    }
+
+    @Test
+    void testClientFailedOfServerNotExists() {
+        Assertions.assertNotNull(Assertions.assertThrows(NotfoundAutoRecoveryException.class, () -> {
+            try (AutoRecoveryClient ignoreClient = AutoRecovery.initClient(taskId, inspectTaskId, tapdataRecoveryEvent -> {
+            })) {
+                // do nothing
+            }
+        }), "Auto-recovery detected");
+    }
+
+    @Test
+    void testDuplicateClient() {
+        Assertions.assertNotNull(Assertions.assertThrows(DuplicateClientAutoRecoveryException.class, () -> {
+            try (
+                AutoRecovery ignoreServer = AutoRecovery.init(taskId);
+                AutoRecoveryClient ignoreClient1 = AutoRecovery.initClient(taskId, inspectTaskId, tapdataRecoveryEvent -> {
+                });
+                AutoRecoveryClient ignoreClient2 = AutoRecovery.initClient(taskId, inspectTaskId, tapdataRecoveryEvent -> {
+                })
+            ) {
+                // do nothing
+            }
+        }), "Duplicate auto-recovery client detected");
+    }
+
+    @Test
+    void testMainFlow() {
+        String inspectTaskId1 = "test-inspect-task-id-1";
+        String inspectTaskId2 = "test-inspect-task-id-2";
+        AtomicBoolean exited = new AtomicBoolean(false);
+        CountDownLatch afterServerInitialized = new CountDownLatch(1);
+        List<Throwable> errors = new ArrayList<>();
+
+        CompletableFuture<Void> serverFuture = CompletableFuture.runAsync(() -> {
+            try (AutoRecovery ignoreAutoRecovery = AutoRecovery.init(taskId)) {
+                ConcurrentLinkedQueue<TapdataRecoveryEvent> queue = new ConcurrentLinkedQueue<>();
+
+                // 初始化 enqueue 后才能初始化 client
+                AutoRecovery.setEnqueueConsumer(taskId, tapdataRecoveryEvent -> {
+                    try {
+                        while (!queue.offer(tapdataRecoveryEvent) && !exited.get()) {
+                            TimeUnit.MILLISECONDS.sleep(interval);
+                        }
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+                afterServerInitialized.countDown();
+
+                // 消费到 end 事件后退出
+                boolean inspectTaskId1Completed = false;
+                boolean inspectTaskId2Completed = false;
+                while (!Thread.currentThread().isInterrupted() && !exited.get()) {
+                    TapdataRecoveryEvent event = queue.poll();
+                    if (event != null) {
+                        System.out.printf("Accept recovery record '%s': %s\n", event.getInspectTaskId(), event.getRecoveryType());
+                        AutoRecovery.completed(taskId, event);
+                        if (TapdataRecoveryEvent.RECOVERY_TYPE_END.equals(event.getRecoveryType())) {
+                            if (inspectTaskId1.equals(event.getInspectTaskId())) {
+                                inspectTaskId1Completed = true;
+                            } else if (inspectTaskId2.equals(event.getInspectTaskId())) {
+                                inspectTaskId2Completed = true;
+                            }
+
+                            if (inspectTaskId1Completed && inspectTaskId2Completed) {
+                                break;
+                            }
+                        }
+                    } else {
+                        TimeUnit.MILLISECONDS.sleep(interval);
+                    }
+                }
+            } catch (Exception e) {
+                errors.add(e);
+            }
+        });
+
+        try {
+            // 验证一个同步任务多校验任务场景
+            CompletableFuture<Void> allFuture = CompletableFuture.allOf(serverFuture
+                , startClient(taskId, inspectTaskId1, afterServerInitialized, errors)
+                , startClient(taskId, inspectTaskId2, afterServerInitialized, errors)
+            );
+            allFuture.get(5, TimeUnit.SECONDS);
+
+            if (!errors.isEmpty()) {
+                Assertions.fail(errors.get(0));
+            }
+        } catch (Exception e) {
+            exited.set(true);
+            Assertions.assertNull(e);
+        }
+    }
+
+    private static CompletableFuture<Void> startClient(String taskId, String inspectTaskId, CountDownLatch afterServerInitialized, List<Throwable> errors) {
+        return CompletableFuture.runAsync(() -> {
+            try {
+                afterServerInitialized.await(2, TimeUnit.SECONDS);
+
+                CountDownLatch completed = new CountDownLatch(1);
+                try (AutoRecoveryClient recoveryClient = AutoRecovery.initClient(taskId, inspectTaskId, tapdataRecoveryEvent -> {
+                    if (TapdataRecoveryEvent.RECOVERY_TYPE_END.equals(tapdataRecoveryEvent.getRecoveryType())) {
+                        completed.countDown();
+                    }
+                })) {
+                    recoveryClient.enqueue(TapdataRecoveryEvent.createBegin(inspectTaskId));
+
+                    String tableId = inspectTaskId + "-table-id";
+                    Map<String, Object> data = new HashMap<>();
+                    recoveryClient.enqueue(TapdataRecoveryEvent.createInsert(inspectTaskId, tableId, data));
+
+                    recoveryClient.enqueue(TapdataRecoveryEvent.createEnd(inspectTaskId));
+
+                    completed.await(5, TimeUnit.SECONDS);
+                }
+            } catch (Exception e) {
+                errors.add(e);
+            }
+        });
+    }
+
+}
diff --git a/iengine/api/src/test/java/io/tapdata/task/TaskSessionClassHolderTest.java b/iengine/api/src/test/java/io/tapdata/task/TaskSessionClassHolderTest.java
new file mode 100644
index 0000000000..f5ee96cce4
--- /dev/null
+++ b/iengine/api/src/test/java/io/tapdata/task/TaskSessionClassHolderTest.java
@@ -0,0 +1,29 @@
+package io.tapdata.task;
+
+import io.tapdata.aspect.TaskStopAspect;
+import io.tapdata.aspect.task.impl.TaskSessionClassHolder;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.mockito.Mockito.*;
+
+public class TaskSessionClassHolderTest {
+
+    @Test
+    void testEnsureTaskSessionStopped(){
+
+        TaskSessionClassHolder taskSessionClassHolder = spy(new TaskSessionClassHolder());
+        TaskStopAspect taskStopAspect = new TaskStopAspect();
+        Map aspectTaskMap = spy(ConcurrentHashMap.class);
+        aspectTaskMap.put("66618104c016c632a6612322",taskStopAspect);
+        String taskId = "66618104c016c632a6612325";
+        ReflectionTestUtils.setField(taskSessionClassHolder,"aspectTaskMap",aspectTaskMap);
+        taskSessionClassHolder.ensureTaskSessionStopped(taskId,taskStopAspect);
+        Assertions.assertTrue(aspectTaskMap.size()==1);
+
+    }
+}
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
index ae9bfa0232..b0c0d6aa0c 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
@@ -200,6 +200,65 @@ void testInitDisableNode() {
 			verify(mockHazelcastBaseNode, new Times(1)).doInitWithDisableNode(jetContext);
 		}
 
+		@Test
+		@SneakyThrows
+		@DisplayName("Init when disable node test")
+		void testInitDisableNodeTestRunTask() {
+			Map<String, Object> attrs = new HashMap<>();
+			attrs.put("disabled", true);
+			tableNode.setAttrs(attrs);
+			TaskDto task = mock(TaskDto.class);
+			when(processorBaseContext.getTaskDto()).thenReturn(task);
+			DAG dag = mock(DAG.class);
+			when(task.getDag()).thenReturn(dag);
+			when(dag.getSourceNode()).thenReturn(new LinkedList<>());
+			when(task.getSyncType()).thenReturn("testRun");
+			doCallRealMethod().when(mockHazelcastBaseNode).init(jetContext);
+			mockHazelcastBaseNode.init(jetContext);
+			verify(mockHazelcastBaseNode, new Times(1)).doInit(jetContext);
+			verify(mockHazelcastBaseNode, new Times(0)).doInitWithDisableNode(jetContext);
+		}
+
+		@Test
+		@SneakyThrows
+		@DisplayName("Init when disable node test")
+		void testInitDisableNodedDeduceSchemaTask() {
+			Map<String, Object> attrs = new HashMap<>();
+			attrs.put("disabled", true);
+			tableNode.setAttrs(attrs);
+			TaskDto task = mock(TaskDto.class);
+			when(processorBaseContext.getTaskDto()).thenReturn(task);
+			DAG dag = mock(DAG.class);
+			when(task.getDag()).thenReturn(dag);
+			when(dag.getSourceNode()).thenReturn(new LinkedList<>());
+			when(task.getSyncType()).thenReturn("deduceSchema");
+			doCallRealMethod().when(mockHazelcastBaseNode).init(jetContext);
+			mockHazelcastBaseNode.init(jetContext);
+			verify(mockHazelcastBaseNode, new Times(1)).doInit(jetContext);
+			verify(mockHazelcastBaseNode, new Times(0)).doInitWithDisableNode(jetContext);
+		}
+
+		@Test
+		@SneakyThrows
+		@DisplayName("Init when disable node test")
+		void testInitDisableNodedSyncTask() {
+			Map<String, Object> attrs = new HashMap<>();
+			attrs.put("disabled", true);
+			tableNode.setAttrs(attrs);
+			TaskDto task = mock(TaskDto.class);
+			when(processorBaseContext.getTaskDto()).thenReturn(task);
+			DAG dag = mock(DAG.class);
+			when(task.getDag()).thenReturn(dag);
+			when(dag.getSourceNode()).thenReturn(new LinkedList<>());
+			when(task.getSyncType()).thenReturn("sync");
+			doCallRealMethod().when(mockHazelcastBaseNode).init(jetContext);
+			mockHazelcastBaseNode.init(jetContext);
+			verify(mockHazelcastBaseNode, new Times(0)).doInit(jetContext);
+			verify(mockHazelcastBaseNode, new Times(1)).doInitWithDisableNode(jetContext);
+		}
+
+
+
 		@Test
 		@SneakyThrows
 		@DisplayName("Init occur error test")
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
index a6c284a428..6b73359219 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
@@ -2,10 +2,13 @@
 
 import base.hazelcast.BaseHazelcastNodeTest;
 import com.google.common.collect.Lists;
+import com.hazelcast.jet.core.JobStatus;
 import com.hazelcast.jet.core.Processor;
+import com.tapdata.entity.*;
 import com.tapdata.entity.Connections;
 import com.tapdata.entity.TapdataEvent;
 import com.tapdata.entity.TapdataShareLogEvent;
+import com.tapdata.entity.task.ExistsDataProcessEnum;
 import com.tapdata.entity.task.context.DataProcessorContext;
 import com.tapdata.tm.commons.dag.DAG;
 import com.tapdata.tm.commons.dag.DmlPolicy;
@@ -16,18 +19,25 @@
 import com.tapdata.tm.commons.dag.process.MergeTableNode;
 import com.tapdata.tm.commons.dag.process.UnwindProcessNode;
 import com.tapdata.tm.commons.task.dto.TaskDto;
+import io.tapdata.aspect.CreateTableFuncAspect;
+import io.tapdata.aspect.DropTableFuncAspect;
 import io.tapdata.entity.codec.filter.TapCodecsFilterManager;
 import io.tapdata.entity.event.TapEvent;
+import io.tapdata.entity.event.ddl.table.TapCreateTableEvent;
 import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import io.tapdata.entity.schema.TapField;
 import io.tapdata.entity.schema.TapTable;
+import io.tapdata.error.TapdataEventException;
 import io.tapdata.flow.engine.V2.exactlyonce.ExactlyOnceUtil;
 import io.tapdata.flow.engine.V2.exactlyonce.write.CheckExactlyOnceWriteEnableResult;
 import io.tapdata.flow.engine.V2.exactlyonce.write.ExactlyOnceWriteCleanerEntity;
+import io.tapdata.flow.engine.V2.monitor.impl.JetJobStatusMonitor;
+import io.tapdata.flow.engine.V2.node.hazelcast.HazelcastBaseNode;
 import io.tapdata.metric.collector.ISyncMetricCollector;
+import io.tapdata.metric.collector.SyncMetricCollector;
 import io.tapdata.observable.logging.ObsLogger;
 import io.tapdata.pdk.apis.entity.Capability;
 import io.tapdata.pdk.apis.entity.ConnectionOptions;
@@ -37,10 +47,12 @@
 import io.tapdata.pdk.apis.functions.connector.target.CreateIndexFunction;
 import io.tapdata.pdk.apis.functions.connector.target.CreateTableFunction;
 import io.tapdata.pdk.apis.functions.connector.target.CreateTableV2Function;
+import io.tapdata.pdk.apis.functions.connector.target.DropTableFunction;
 import io.tapdata.pdk.core.api.ConnectorNode;
 import io.tapdata.pdk.core.async.AsyncUtils;
 import io.tapdata.pdk.core.async.ThreadPoolExecutorEx;
 import io.tapdata.pdk.core.utils.CommonUtils;
+import io.tapdata.utils.UnitTestUtils;
 import org.junit.jupiter.api.*;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
@@ -54,6 +66,7 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -246,7 +259,7 @@ void testIsUnwindProcess() {
 			hazelcastTargetPdkBaseNode.unwindProcess = true;
 			TapTable tapTable = mock(TapTable.class);
 			AtomicBoolean atomicBoolean = new AtomicBoolean(false);
-			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable, atomicBoolean);
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable, atomicBoolean,true);
 			TableNode node = new TableNode();
 			node.setDisabled(false);
 			when(hazelcastTargetPdkBaseNode.getNode()).thenReturn((Node) node);
@@ -259,7 +272,7 @@ void testIsUnwindProcess() {
 			Connections connections = new Connections();
 			connections.setId("test");
 			when(dataProcessorContext.getTargetConn()).thenReturn(connections);
-			boolean result = hazelcastTargetPdkBaseNode.createTable(tapTable, atomicBoolean);
+			boolean result = hazelcastTargetPdkBaseNode.createTable(tapTable, atomicBoolean,true);
 			Assertions.assertTrue(result);
 		}
 
@@ -268,7 +281,7 @@ void testIsNotUnwindProcess() {
 			hazelcastTargetPdkBaseNode.unwindProcess = false;
 			TapTable tapTable = mock(TapTable.class);
 			AtomicBoolean atomicBoolean = new AtomicBoolean(false);
-			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable, atomicBoolean);
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable, atomicBoolean,true);
 			TableNode node = new TableNode();
 			node.setDisabled(false);
 			when(hazelcastTargetPdkBaseNode.getNode()).thenReturn((Node) node);
@@ -281,7 +294,7 @@ void testIsNotUnwindProcess() {
 			Connections connections = new Connections();
 			connections.setId("test");
 			when(dataProcessorContext.getTargetConn()).thenReturn(connections);
-			boolean result = hazelcastTargetPdkBaseNode.createTable(tapTable, atomicBoolean);
+			boolean result = hazelcastTargetPdkBaseNode.createTable(tapTable, atomicBoolean,true);
 			Assertions.assertTrue(result);
 		}
 		@Test
@@ -300,8 +313,8 @@ void testCreateTableIsNull(){
 			when(functions.getCreateTableFunction()).thenReturn(null);
 			when(functions.getCreateTableV2Function()).thenReturn(null);
 			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
-			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable,succeed);
-			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed);
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable,succeed,true);
+			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed,true);
 			verify(hazelcastTargetPdkBaseNode,new Times(0)).buildErrorConsumer("test");
 		}
 		@Test
@@ -320,10 +333,73 @@ void testCreateTableForBuildErrorConsumer(){
 			when(functions.getCreateTableFunction()).thenReturn(mock(CreateTableFunction.class));
 			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
 			doCallRealMethod().when(hazelcastTargetPdkBaseNode).executeDataFuncAspect(any(Class.class),any(Callable.class),any(CommonUtils.AnyErrorConsumer.class));
-			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable,succeed);
-			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed);
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable,succeed,true);
+			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed,true);
 			verify(hazelcastTargetPdkBaseNode,new Times(1)).buildErrorConsumer("test");
 		}
+
+		@Test
+		void createTableTestForInit(){
+			ReflectionTestUtils.setField(hazelcastTargetPdkBaseNode,"unwindProcess",false);
+			TapTable tapTable = new TapTable();
+			tapTable.setId("test");
+			AtomicBoolean succeed = new AtomicBoolean(true);
+			Node node = mock(Node.class);
+			when(hazelcastTargetPdkBaseNode.getNode()).thenReturn(node);
+			when(node.disabledNode()).thenReturn(false);
+			ConnectorNode connectorNode = mock(ConnectorNode.class);
+			when(hazelcastTargetPdkBaseNode.getConnectorNode()).thenReturn(connectorNode);
+			ConnectorFunctions functions = mock(ConnectorFunctions.class);
+			when(connectorNode.getConnectorFunctions()).thenReturn(functions);
+			when(functions.getCreateTableFunction()).thenReturn(mock(CreateTableFunction.class));
+			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
+			when(hazelcastTargetPdkBaseNode.executeDataFuncAspect(any(Class.class), any(Callable.class), any(CommonUtils.AnyErrorConsumer.class))).thenAnswer(a -> {
+				Callable<?> callable = a.getArgument(1, Callable.class);
+				CommonUtils.AnyErrorConsumer<CreateTableFuncAspect> errorConsumer = a.getArgument(2, CommonUtils.AnyErrorConsumer.class);
+				Object call = callable.call();
+
+				Assertions.assertNotNull(call);
+				Assertions.assertTrue(((CreateTableFuncAspect)call).isInit());
+                return null;
+            });
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable,succeed,true);
+			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed,true);
+		}
+
+		@Test
+		void dropTableTestForInit(){
+			TaskDto taskDto = new TaskDto();
+			taskDto.setType("initial_sync");
+			when(dataProcessorContext.getTaskDto()).thenReturn(taskDto);
+			HazelcastTargetPdkDataNode	hazelcastTargetPdkDataNode = mock(HazelcastTargetPdkDataNode.class);
+			ReflectionTestUtils.setField(hazelcastTargetPdkDataNode, "dataProcessorContext", dataProcessorContext);
+			ReflectionTestUtils.setField(hazelcastTargetPdkDataNode, "clientMongoOperator", mockClientMongoOperator);
+			TapTable tapTable = new TapTable();
+			tapTable.setId("test");
+			Node node = mock(Node.class);
+			when(hazelcastTargetPdkDataNode.getNode()).thenReturn(node);
+			when(node.disabledNode()).thenReturn(false);
+			ConnectorNode connectorNode = mock(ConnectorNode.class);
+			when(hazelcastTargetPdkDataNode.getConnectorNode()).thenReturn(connectorNode);
+			ConnectorFunctions functions = mock(ConnectorFunctions.class);
+			when(connectorNode.getConnectorFunctions()).thenReturn(functions);
+			when(functions.getDropTableFunction()).thenReturn(mock(DropTableFunction.class));
+			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
+
+
+			when(hazelcastTargetPdkDataNode.executeDataFuncAspect(any(Class.class), any(Callable.class), any(CommonUtils.AnyErrorConsumer.class))).thenAnswer(a -> {
+				Callable<?> callable = a.getArgument(1, Callable.class);
+				CommonUtils.AnyErrorConsumer<DropTableFuncAspect> errorConsumer = a.getArgument(2, CommonUtils.AnyErrorConsumer.class);
+				Object call = callable.call();
+
+				Assertions.assertNotNull(call);
+				Assertions.assertTrue(((DropTableFuncAspect)call).isInit());
+				return null;
+			});
+			ExistsDataProcessEnum existsDataProcessEnum = ExistsDataProcessEnum.DROP_TABLE;
+			doCallRealMethod().when(hazelcastTargetPdkDataNode).dropTable(existsDataProcessEnum,"test",true);
+			hazelcastTargetPdkDataNode.dropTable(existsDataProcessEnum, "test",true);
+		}
 	}
 
 	@Nested
@@ -795,7 +871,7 @@ void test1(){
 				when(checkExactlyOnceWriteEnableResult.getEnable()).thenReturn(true);
 				ConnectorFunctions functions = mock(ConnectorFunctions.class);
 				when(connectorNode.getConnectorFunctions()).thenReturn(functions);
-				when(hazelcastTargetPdkBaseNode.createTable(any(TapTable.class), any(AtomicBoolean.class))).thenReturn(true);
+				when(hazelcastTargetPdkBaseNode.createTable(any(TapTable.class), any(AtomicBoolean.class),any(Boolean.class))).thenReturn(true);
 				when(functions.getCreateIndexFunction()).thenReturn(mock(CreateIndexFunction.class));
 				when(hazelcastTargetPdkBaseNode.getNode()).thenReturn((Node) mock(TableNode.class));
 				doCallRealMethod().when(hazelcastTargetPdkBaseNode).initExactlyOnceWriteIfNeed();
@@ -804,4 +880,215 @@ void test1(){
 			}
 		}
 	}
+
+    @Nested
+    class HandleTapdataEventsTest {
+        List<TapdataEvent> tapdataEvents = new ArrayList<>();
+        JetJobStatusMonitor jobStatusMonitor = mock(JetJobStatusMonitor.class);
+
+        @BeforeEach
+        void setUp(){
+            doCallRealMethod().when(hazelcastTargetPdkBaseNode).handleTapdataEvents(any());
+            when(jobStatusMonitor.get()).thenReturn(JobStatus.RUNNING);
+
+            UnitTestUtils.injectField(HazelcastBaseNode.class, hazelcastTargetPdkBaseNode, "running", new AtomicBoolean(true));
+            UnitTestUtils.injectField(HazelcastBaseNode.class, hazelcastTargetPdkBaseNode, "jetJobStatusMonitor", jobStatusMonitor);
+            UnitTestUtils.injectField(HazelcastTargetPdkBaseNode.class, hazelcastTargetPdkBaseNode, "firstStreamEvent", new AtomicBoolean(false));
+            UnitTestUtils.injectField(HazelcastTargetPdkBaseNode.class, hazelcastTargetPdkBaseNode, "exactlyOnceWriteNeedLookupTables", new ConcurrentHashMap<>());
+            when(hazelcastTargetPdkBaseNode.getConnectorNode()).thenReturn(mock(ConnectorNode.class));
+        }
+
+        @Test
+        void testEmptyEvents() {
+            hazelcastTargetPdkBaseNode.handleTapdataEvents(tapdataEvents);
+            verify(hazelcastTargetPdkBaseNode, times(0)).handleTapdataEvent(any(), any(), any(), any(), any(), any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).processTapEvents(any(), any(), any());
+        }
+
+        @Test
+        void testRunning() {
+            // running false
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            TapdataEvent tapdataEvent = new TapdataEvent();
+            tapdataEvent.setTapEvent(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+            tapdataEvents.add(tapdataEvent);
+            hazelcastTargetPdkBaseNode.handleTapdataEvents(tapdataEvents);
+
+            // running true
+            when(hazelcastTargetPdkBaseNode.isRunning()).thenReturn(true);
+            hazelcastTargetPdkBaseNode.handleTapdataEvents(tapdataEvents);
+            verify(hazelcastTargetPdkBaseNode, times(1)).handleTapdataEvent(any(), any(), any(), any(), any(), any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).processTapEvents(any(), any(), any());
+        }
+
+        @Test
+        void testShareLog() {
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            TapdataShareLogEvent shareLogEvent = new TapdataShareLogEvent();
+            shareLogEvent.setSyncStage(SyncStage.CDC);
+            shareLogEvent.setTapEvent(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+            tapdataEvents.add(shareLogEvent);
+
+            hazelcastTargetPdkBaseNode.syncMetricCollector = mock(SyncMetricCollector.class);
+            when(hazelcastTargetPdkBaseNode.isRunning()).thenReturn(true);
+            doAnswer(invocationOnMock -> {
+                invocationOnMock.getArgument(1, List.class).add(invocationOnMock.getArgument(5));
+                return null;
+            }).when(hazelcastTargetPdkBaseNode).handleTapdataEvent(any(), any(), any(), any(), any(), any());
+
+            hazelcastTargetPdkBaseNode.handleTapdataEvents(tapdataEvents);
+            verify(hazelcastTargetPdkBaseNode, times(1)).handleTapdataEvent(any(), any(), any(), any(), any(), any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).processShareLog(any());
+        }
+
+        @Test
+        void testProcessError() {
+            tapdataEvents.add(TapdataStartedCdcEvent.create());
+
+            doAnswer(invocationOnMock -> {
+                throw new RuntimeException("test");
+            }).when(hazelcastTargetPdkBaseNode).processTapEvents(any(), any(), any());
+
+            when(hazelcastTargetPdkBaseNode.isRunning()).thenReturn(true);
+            Assertions.assertThrows(TapdataEventException.class, () -> hazelcastTargetPdkBaseNode.handleTapdataEvents(tapdataEvents));
+        }
+    }
+
+    @Nested
+    class HandleTapdataEventTest {
+        List<TapEvent> tapEvents = new ArrayList<>();
+        AtomicBoolean hasExactlyOnceWriteCache = new AtomicBoolean(false);
+        List<TapRecordEvent> exactlyOnceWriteCache = new ArrayList<>();
+        AtomicReference<TapdataEvent> lastTapdataEvent = new AtomicReference<>();
+
+        @BeforeEach
+        void setUp() throws Exception {
+            doCallRealMethod().when(hazelcastTargetPdkBaseNode).handleTapdataEvent(any(), any(), any(), any(), any());
+        }
+
+        @Test
+        void testDDL() {
+            TapdataEvent tapdataEvent = new TapdataEvent();
+            tapdataEvent.setTapEvent(new TapCreateTableEvent());
+
+            assertDoesNotThrow(() -> hazelcastTargetPdkBaseNode.handleTapdataEvent(tapEvents, hasExactlyOnceWriteCache, exactlyOnceWriteCache, lastTapdataEvent, tapdataEvent));
+        }
+
+        @Test
+        void testDML() {
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            TapdataEvent tapdataEvent = new TapdataEvent();
+            tapdataEvent.setTapEvent(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+
+            assertDoesNotThrow(() -> hazelcastTargetPdkBaseNode.handleTapdataEvent(tapEvents, hasExactlyOnceWriteCache, exactlyOnceWriteCache, lastTapdataEvent, tapdataEvent));
+        }
+
+        @Test
+        void testOther() {
+            TapdataEvent tapdataEvent = new TapdataEvent();
+
+            assertDoesNotThrow(() -> hazelcastTargetPdkBaseNode.handleTapdataEvent(tapEvents, hasExactlyOnceWriteCache, exactlyOnceWriteCache, lastTapdataEvent, tapdataEvent));
+        }
+    }
+
+    @Nested
+    class ProcessTapEventsTest {
+        List<TapdataEvent> tapdataEvents;
+        List<TapEvent> tapEvents;
+        CheckExactlyOnceWriteEnableResult checkExactlyOnceWriteEnableResult;
+        AtomicBoolean hasExactlyOnceWriteCache;
+
+        @BeforeEach
+        void setUp() {
+            tapdataEvents = new ArrayList<>();
+            tapEvents = new ArrayList<>();
+            checkExactlyOnceWriteEnableResult = CheckExactlyOnceWriteEnableResult.createEnable();
+            hasExactlyOnceWriteCache = new AtomicBoolean(false);
+            doCallRealMethod().when(hazelcastTargetPdkBaseNode).processTapEvents(any(), any(), any());
+            UnitTestUtils.injectField(HazelcastTargetPdkBaseNode.class, hazelcastTargetPdkBaseNode, "checkExactlyOnceWriteEnableResult", checkExactlyOnceWriteEnableResult);
+        }
+
+        @Test
+        void testEmpty() {
+            hazelcastTargetPdkBaseNode.processTapEvents(tapdataEvents, tapEvents, hasExactlyOnceWriteCache);
+        }
+
+        @Test
+        void testExactlyOnceWriteCacheFalse() {
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            tapEvents.add(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+            hazelcastTargetPdkBaseNode.processTapEvents(tapdataEvents, tapEvents, hasExactlyOnceWriteCache);
+            verify(hazelcastTargetPdkBaseNode, times(1)).processEvents(any());
+        }
+
+        @Test
+        void testExactlyOnceWriteCacheTrue() {
+            hasExactlyOnceWriteCache.set(true);
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            tapEvents.add(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+            hazelcastTargetPdkBaseNode.processTapEvents(tapdataEvents, tapEvents, hasExactlyOnceWriteCache);
+            verify(hazelcastTargetPdkBaseNode, times(1)).transactionBegin();
+            verify(hazelcastTargetPdkBaseNode, times(1)).processEvents(any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).processExactlyOnceWriteCache(any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).transactionCommit();
+        }
+
+        @Test
+        void testRollback() {
+            hasExactlyOnceWriteCache.set(true);
+            String testTableName = "test-table";
+            Map<String, Object> testAfterData = new HashMap<>();
+            tapEvents.add(TapInsertRecordEvent.create().table(testTableName).after(testAfterData));
+
+            doAnswer(invocationOnMock -> {
+                throw new RuntimeException("test");
+            }).when(hazelcastTargetPdkBaseNode).processEvents(any());
+
+            assertThrows(RuntimeException.class, () -> {
+                hazelcastTargetPdkBaseNode.processTapEvents(tapdataEvents, tapEvents, hasExactlyOnceWriteCache);
+            });
+            verify(hazelcastTargetPdkBaseNode, times(1)).transactionBegin();
+            verify(hazelcastTargetPdkBaseNode, times(1)).processEvents(any());
+            verify(hazelcastTargetPdkBaseNode, times(1)).transactionRollback();
+        }
+    }
+
+    @Nested
+    class HandleAspectWithSyncStageTest {
+
+        AtomicBoolean firstBatchEvent;
+        AtomicBoolean firstStreamEvent;
+
+        @BeforeEach
+        void setUp() {
+            firstBatchEvent = new AtomicBoolean(false);
+            firstStreamEvent = new AtomicBoolean(false);
+            UnitTestUtils.injectField(HazelcastTargetPdkBaseNode.class, hazelcastTargetPdkBaseNode, "firstBatchEvent", firstBatchEvent);
+            UnitTestUtils.injectField(HazelcastTargetPdkBaseNode.class, hazelcastTargetPdkBaseNode, "firstStreamEvent", firstStreamEvent);
+            hazelcastTargetPdkBaseNode.syncMetricCollector = mock(SyncMetricCollector.class);
+            doCallRealMethod().when(hazelcastTargetPdkBaseNode).handleAspectWithSyncStage(any());
+        }
+
+        @Test
+        void testInitialSync() {
+            hazelcastTargetPdkBaseNode.handleAspectWithSyncStage(SyncStage.INITIAL_SYNC);
+            hazelcastTargetPdkBaseNode.handleAspectWithSyncStage(SyncStage.INITIAL_SYNC);
+            assertTrue(firstBatchEvent.get());
+            assertFalse(firstStreamEvent.get());
+        }
+
+        @Test
+        void testCdc() {
+            hazelcastTargetPdkBaseNode.handleAspectWithSyncStage(SyncStage.CDC);
+            hazelcastTargetPdkBaseNode.handleAspectWithSyncStage(SyncStage.CDC);
+            assertFalse(firstBatchEvent.get());
+            assertTrue(firstStreamEvent.get());
+        }
+    }
+
 }
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNodeTest.java
index 10c59614c9..301d524937 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNodeTest.java
@@ -124,20 +124,20 @@ void beforeEach(){
 			node = mock(TableNode.class);
 			existsDataProcessEnum = mock(ExistsDataProcessEnum.class);
 			tableId = "test";
-			doCallRealMethod().when(hazelcastTargetPdkDataNode).createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId);
+			doCallRealMethod().when(hazelcastTargetPdkDataNode).createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId,true);
 		}
 		@Test
 		@DisplayName("test createTable method when tapTable is null")
 		void testCreateTable1(){
-			assertThrows(TapCodeException.class, ()->hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId));
+			assertThrows(TapCodeException.class, ()->hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId,true));
 		}
 		@Test
 		@DisplayName("test createTable method when tableId equals TAP_EXACTLY_ONCE_CACHE")
 		void testCreateTable2(){
 			tableId = "_TAP_EXACTLY_ONCE_CACHE";
 			when(tapTableMap.get(tableId)).thenReturn(mock(TapTable.class));
-			doCallRealMethod().when(hazelcastTargetPdkDataNode).createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId);
-			hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId);
+			doCallRealMethod().when(hazelcastTargetPdkDataNode).createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId,true);
+			hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId,true);
 			verify(hazelcastTargetPdkDataNode,new Times(0)).syncIndex(anyString(),any(TapTable.class),anyBoolean());
 		}
 		@Test
@@ -146,7 +146,7 @@ void testCreateTable3(){
 			when(tapTableMap.get(tableId)).thenReturn(mock(TapTable.class));
 			ReflectionTestUtils.setField(hazelcastTargetPdkDataNode,"writeStrategy","appendWrite");
 			when(funcAspect.state(TableInitFuncAspect.STATE_PROCESS)).thenReturn(mock(TableInitFuncAspect.class));
-			hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId);
+			hazelcastTargetPdkDataNode.createTable(tapTableMap,funcAspect,node,existsDataProcessEnum,tableId,true);
 			verify(hazelcastTargetPdkDataNode,new Times(1)).syncIndex(anyString(),any(TapTable.class),anyBoolean());
 		}
 	}
@@ -637,8 +637,8 @@ void test1(){
 			when(connectorNode.getConnectorFunctions()).thenReturn(functions);
 			when(functions.getDropTableFunction()).thenReturn(mock(DropTableFunction.class));
 			doCallRealMethod().when(hazelcastTargetPdkDataNode).executeDataFuncAspect(any(Class.class),any(Callable.class),any(CommonUtils.AnyErrorConsumer.class));
-			doCallRealMethod().when(hazelcastTargetPdkDataNode).dropTable(existsDataProcessEnum,tableId);
-			hazelcastTargetPdkDataNode.dropTable(existsDataProcessEnum,tableId);
+			doCallRealMethod().when(hazelcastTargetPdkDataNode).dropTable(existsDataProcessEnum,tableId,true);
+			hazelcastTargetPdkDataNode.dropTable(existsDataProcessEnum,tableId,true);
 			verify(hazelcastTargetPdkDataNode,new Times(1)).buildErrorConsumer(tableId);
 		}
 	}
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJoinProcessorTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJoinProcessorTest.java
new file mode 100644
index 0000000000..e5730a0641
--- /dev/null
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJoinProcessorTest.java
@@ -0,0 +1,65 @@
+package io.tapdata.flow.engine.V2.node.hazelcast.processor;
+
+import base.hazelcast.BaseHazelcastNodeTest;
+import com.tapdata.tm.commons.dag.process.JoinProcessorNode;
+import io.tapdata.entity.codec.filter.impl.AllLayerMapIterator;
+import io.tapdata.entity.schema.TapTable;
+import io.tapdata.entity.schema.value.DateTime;
+import io.tapdata.flow.engine.V2.node.hazelcast.processor.join.HazelcastJoinProcessor;
+import io.tapdata.schema.TapTableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+
+import static org.mockito.Mockito.*;
+
+
+public class HazelcastJoinProcessorTest extends BaseHazelcastNodeTest {
+    HazelcastJoinProcessor hazelcastJoinProcessor;
+    JoinProcessorNode joinProcessorNode;
+    @BeforeEach
+    void beforeEach() {
+        super.allSetup();
+        joinProcessorNode = new JoinProcessorNode();
+        joinProcessorNode.setId("node_test");
+        joinProcessorNode.setJoinType("left");
+        joinProcessorNode.setEmbeddedMode(false);
+        JoinProcessorNode.JoinExpression joinExpression = new JoinProcessorNode.JoinExpression();
+        joinExpression.setExpression("test");
+        joinExpression.setLeft("left");
+        joinExpression.setRight("right");
+        joinProcessorNode.setJoinExpressions(Arrays.asList(joinExpression));
+        joinProcessorNode.setLeftNodeId("leftNodeId");
+        joinProcessorNode.setRightNodeId("rightNodeId");
+        joinProcessorNode.setLeftPrimaryKeys(new ArrayList<>());
+        joinProcessorNode.setRightPrimaryKeys(new ArrayList<>());
+        TapTableMap<String, TapTable> tapTableMap = mock(TapTableMap.class);
+        when(processorBaseContext.getTapTableMap()).thenReturn(tapTableMap);
+        when(tapTableMap.keySet()).thenReturn(new HashSet<>(Arrays.asList("left")));
+        when(tapTableMap.get("left")).thenReturn(mock(TapTable.class));
+        hazelcastJoinProcessor = spy(new HazelcastJoinProcessor(processorBaseContext));
+    }
+
+    @Test
+    void testDoInit(){
+        doNothing().when(hazelcastJoinProcessor).initNode();
+        hazelcastJoinProcessor.doInit(jetContext);
+        Assertions.assertNotNull(ReflectionTestUtils.getField(hazelcastJoinProcessor,"mapIterator"));
+    }
+    @Test
+    void testTransformDateTime(){
+        ReflectionTestUtils.setField(hazelcastJoinProcessor,"mapIterator",new AllLayerMapIterator());
+        Map<String, Object> before = new HashMap<>();
+        before.put("date",new DateTime(new Date()));
+        before.put("text","text");
+        Map<String, Object> after = new HashMap<>();
+        after.put("date",new DateTime(new Date()));
+        after.put("text","text");
+        hazelcastJoinProcessor.transformDateTime(before,after);
+        Assertions.assertInstanceOf(Date.class, before.get("date"));
+        Assertions.assertInstanceOf(Date.class, after.get("date"));
+    }
+}
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 bcc9f33b4f..baa70f51ad 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
@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.jet.core.Processor;
+import com.tapdata.entity.task.context.ProcessorBaseContext;
 import io.tapdata.entity.schema.value.TapArrayValue;
 import io.tapdata.entity.schema.value.TapMapValue;
 import io.tapdata.utils.AppType;
@@ -342,7 +343,7 @@ void transformDateTime() {
 			}});
 			hazelcastMergeNode.transformDateTime(after);
 			assertEquals(4, after.size());
-			assertInstanceOf(Date.class, after.get("create_time"));
+			assertInstanceOf(Instant.class, after.get("create_time"));
 		}
 	}
 
@@ -1719,12 +1720,28 @@ void test2() {
 			TapdataEvent tapdataEvent = new TapdataEvent();
 			tapdataEvent.addNodeId("123");
 			tapdataEvent.setSyncStage(SyncStage.INITIAL_SYNC);
-			TapUpdateRecordEvent tapUpdateRecordEvent = TapUpdateRecordEvent.create().init();
-			tapdataEvent.setTapEvent(tapUpdateRecordEvent);
+			TapInsertRecordEvent tapInsertRecordEvent = TapInsertRecordEvent.create().init();
+			tapdataEvent.setTapEvent(tapInsertRecordEvent);
 			when(mockHazelcastMergeNode.isSubTableFirstMode()).thenReturn(true);
 			boolean result = mockHazelcastMergeNode.needLookup(tapdataEvent);
 			assertEquals(true, result);
 		}
+
+		@Test
+		@DisplayName("test cdc task, update and delete TapEvent, expect return false")
+		void test3() {
+			processorBaseContext.getTaskDto().setType(SyncTypeEnum.CDC.getSyncType());
+			doReturn(false).when(mockHazelcastMergeNode).isSubTableFirstMode();
+			TapdataEvent tapdataEvent = new TapdataEvent();
+			TapUpdateRecordEvent tapUpdateRecordEvent = TapUpdateRecordEvent.create().init();
+			tapdataEvent.setTapEvent(tapUpdateRecordEvent);
+			assertFalse(mockHazelcastMergeNode.needLookup(tapdataEvent));
+
+			TapDeleteRecordEvent tapDeleteRecordEvent = TapDeleteRecordEvent.create().init();
+			tapdataEvent = new TapdataEvent();
+			tapdataEvent.setTapEvent(tapDeleteRecordEvent);
+			assertFalse(mockHazelcastMergeNode.needLookup(tapdataEvent));
+		}
 	}
 
 	@Nested
@@ -2207,4 +2224,63 @@ void testUpdateWriteLookupMoreThanOneRow() {
 			verify(nodeLogger, times(1)).warn(eq("Update write merge lookup, find more than one row, lookup table: {}, join key value: {}, will use first row: {}"), any(Object[].class));
 		}
 	}
+	@Nested
+	class InitSourcePkOrUniqueFieldMapTest{
+		ProcessorBaseContext processorBaseContextTest;
+		private Map<String, List<String>> sourcePkOrUniqueFieldMap;
+		@BeforeEach
+		void setUp(){
+			sourcePkOrUniqueFieldMap = new HashMap<>();
+			processorBaseContextTest = mock(ProcessorBaseContext.class);
+			ReflectionTestUtils.setField(hazelcastMergeNode, "processorBaseContext", processorBaseContextTest);
+			ReflectionTestUtils.setField(hazelcastMergeNode, "sourcePkOrUniqueFieldMap", sourcePkOrUniqueFieldMap);
+		}
+
+
+		@Test
+		void testNodeIsDisabled(){
+			MergeTableProperties mergeTableProperties = new MergeTableProperties();
+			mergeTableProperties.setId("2cbc1a4d-906d-4b32-9cf4-6596ed4bd0e4");
+			mergeTableProperties.setMergeType(MergeTableProperties.MergeType.updateWrite);
+			mergeTableProperties.setChildren(new ArrayList<>());
+			List<MergeTableProperties> mergeTablePropertiesList  = new ArrayList<>();
+			mergeTablePropertiesList.add(mergeTableProperties);
+			List<Node> nodes = new ArrayList<>();
+			Map<String, Object> attrs = new HashMap<>();
+			attrs.put("disabled",true);
+			TableNode node = new TableNode();
+			node.setId("2cbc1a4d-906d-4b32-9cf4-6596ed4bd0e4");
+			node.setAttrs(attrs);
+			nodes.add(node);
+			when(processorBaseContextTest.getNodes()).thenReturn(nodes);
+			hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
+			Assertions.assertEquals(0,sourcePkOrUniqueFieldMap.size());
+		}
+
+		@Test
+		void testNodeIsNotDisabled(){
+			MergeTableProperties mergeTableProperties = new MergeTableProperties();
+			mergeTableProperties.setId("2cbc1a4d-906d-4b32-9cf4-6596ed4bd0e4");
+			mergeTableProperties.setMergeType(MergeTableProperties.MergeType.updateWrite);
+			mergeTableProperties.setChildren(new ArrayList<>());
+			List<MergeTableProperties> mergeTablePropertiesList  = new ArrayList<>();
+			mergeTablePropertiesList.add(mergeTableProperties);
+			List<Node> nodes = new ArrayList<>();
+			Map<String, Object> attrs = new HashMap<>();
+			TableNode node = new TableNode();
+			node.setId("2cbc1a4d-906d-4b32-9cf4-6596ed4bd0e4");
+			node.setTableName("test");
+			node.setAttrs(attrs);
+			nodes.add(node);
+			when(processorBaseContextTest.getNodes()).thenReturn(nodes);
+			TapTableMap tapTableMap = mock(TapTableMap.class);
+			when(processorBaseContextTest.getTapTableMap()).thenReturn(tapTableMap);
+			TapTable tapTable = mock(TapTable.class);
+			when(tapTableMap.get(any())).thenReturn(tapTable);
+			when(tapTable.primaryKeys(true)).thenReturn(Arrays.asList("id"));
+			hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
+			Assertions.assertEquals(1,sourcePkOrUniqueFieldMap.size());
+		}
+
+	}
 }
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 622df47ad6..908090629a 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
@@ -396,6 +396,68 @@ void testCreateNode21(){
             HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
             assertEquals(HazelcastAddDateFieldProcessNode.class, actual.getClass());
         }
+
+        @Test
+        @SneakyThrows
+        @DisplayName("test createNode method for add date field processor")
+        void testCreateNodeTestRunTask(){
+            node = mock(AddDateFieldProcessorNode.class);
+            when(node.getType()).thenReturn("add_date_field_processor");
+            when(taskDto.getType()).thenReturn("initial_sync");
+            when(taskDto.getSyncType()).thenReturn("testRun");
+            HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
+            assertEquals(HazelcastAddDateFieldProcessNode.class, actual.getClass());
+        }
+
+        @Test
+        @SneakyThrows
+        @DisplayName("test createNode method for add date field processor")
+        void testCreateNodeDeduceSchemaTask(){
+            node = mock(AddDateFieldProcessorNode.class);
+            when(node.getType()).thenReturn("add_date_field_processor");
+            when(taskDto.getType()).thenReturn("initial_sync");
+            when(taskDto.getSyncType()).thenReturn("deduceSchema");
+            HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
+            assertEquals(HazelcastAddDateFieldProcessNode.class, actual.getClass());
+        }
+
+        @Test
+        @SneakyThrows
+        @DisplayName("test createNode method for add date field processor")
+        void testCreateNodeSyncTask(){
+            node = mock(AddDateFieldProcessorNode.class);
+            when(node.getType()).thenReturn("add_date_field_processor");
+            when(node.disabledNode()).thenReturn(true);
+            when(taskDto.getType()).thenReturn("initial_sync");
+            when(taskDto.getSyncType()).thenReturn("sync");
+            HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
+            assertEquals(HazelcastBlank.class, actual.getClass());
+        }
+
+        @Test
+        @SneakyThrows
+        @DisplayName("test createNode method for migrate js processor")
+        void testCreateNodeMigrateTask(){
+            node = mock(MigrateJsProcessorNode.class);
+            when(node.getType()).thenReturn("migrate_js_processor");
+            when(node.disabledNode()).thenReturn(true);
+            when(taskDto.getType()).thenReturn("initial_sync");
+            when(taskDto.getSyncType()).thenReturn("migrate");
+            HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
+            assertEquals(HazelcastBlank.class, actual.getClass());
+        }
+        @Test
+        @SneakyThrows
+        @DisplayName("test createNode method for migrate js processor")
+        void testCreateNodeMigrateTaskDisableIsFalse(){
+            node = mock(MigrateJsProcessorNode.class);
+            when(node.getType()).thenReturn("migrate_js_processor");
+            when(node.disabledNode()).thenReturn(false);
+            when(taskDto.getType()).thenReturn("initial_sync");
+            when(taskDto.getSyncType()).thenReturn("migrate");
+            HazelcastBaseNode actual = HazelcastTaskService.createNode(taskDto, nodes, edges, node, predecessors, successors, config, connection, databaseType, mergeTableMap, tapTableMap, taskConfig);
+            assertEquals(HazelcastJavaScriptProcessorNode.class, actual.getClass());
+        }
     }
 
     @Nested
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/processor/FieldProcessUtilTest.java b/iengine/iengine-common/src/test/java/com/tapdata/processor/FieldProcessUtilTest.java
new file mode 100644
index 0000000000..c9af9acf5f
--- /dev/null
+++ b/iengine/iengine-common/src/test/java/com/tapdata/processor/FieldProcessUtilTest.java
@@ -0,0 +1,57 @@
+package com.tapdata.processor;
+
+import io.tapdata.entity.schema.value.DateTime;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.internal.verification.Times;
+
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class FieldProcessUtilTest {
+    @Nested
+    class convertNumber{
+        private Object value;
+        private Function fn;
+        @BeforeEach
+        void beforeEach(){
+            fn = mock(Function.class);
+        }
+        @Test
+        @DisplayName("test convertNumber method for blank str")
+        void test1(){
+            value = "";
+            FieldProcessUtil.convertNumber(value, fn);
+            verify(fn, new Times(0)).apply(anyString());
+        }
+        @Test
+        @DisplayName("test convertNumber method for int")
+        void test2(){
+            value = 1;
+            FieldProcessUtil.convertNumber(value, fn);
+            verify(fn, new Times(1)).apply(anyString());
+        }
+    }
+    @Nested
+    class handleDateTime{
+        @Test
+        void testForDateTime(){
+            Long expect = 1718953878000L;
+            Object value = new DateTime(expect);
+            Object actual = FieldProcessUtil.handleDateTime(value);
+            assertEquals(expect, actual);
+        }
+        @Test
+        void testForLong(){
+            Object value = 1L;
+            Object actual = FieldProcessUtil.handleDateTime(value);
+            assertEquals(value, actual);
+        }
+    }
+}
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/ObservableAspectTaskTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/ObservableAspectTaskTest.java
index 2b3968004f..251c0d0017 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/ObservableAspectTaskTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/ObservableAspectTaskTest.java
@@ -1,7 +1,10 @@
 package io.tapdata.observable.metric;
 
+import com.tapdata.entity.task.context.DataProcessorContext;
 import com.tapdata.mongo.RestTemplateOperator;
+import com.tapdata.tm.commons.dag.Node;
 import com.tapdata.tm.commons.task.dto.TaskDto;
+import io.tapdata.aspect.DropTableFuncAspect;
 import io.tapdata.aspect.TaskStartAspect;
 import io.tapdata.aspect.TaskStopAspect;
 import io.tapdata.entity.aspect.Aspect;
@@ -9,6 +12,7 @@
 import io.tapdata.observable.metric.handler.DataNodeSampleHandler;
 import io.tapdata.observable.metric.handler.ProcessorNodeSampleHandler;
 import io.tapdata.observable.metric.handler.TableSampleHandler;
+import io.tapdata.observable.metric.handler.TaskSampleHandler;
 import io.tapdata.observable.metric.util.SyncGetMemorySizeHandler;
 import org.bson.types.ObjectId;
 import org.junit.jupiter.api.Assertions;
@@ -18,10 +22,9 @@
 import org.mockito.Mockito;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
+
+import static org.mockito.Mockito.*;
 
 
 public class ObservableAspectTaskTest {
@@ -29,6 +32,9 @@ public class ObservableAspectTaskTest {
     @BeforeEach
     void init() {
         observableAspectTask = new ObservableAspectTask();
+        TaskDto taskDto = new TaskDto();
+        taskDto.setId(ObjectId.get());
+        ReflectionTestUtils.setField(observableAspectTask,"task",taskDto);
     }
 
     @Nested
@@ -53,7 +59,7 @@ void testStartAndStop() {
         taskDto.setStartTime(new Date());
         observableAspectTask.setTask(taskDto);
 
-        RestTemplateOperator mockRestTemplateOperator = Mockito.mock(RestTemplateOperator.class);
+        RestTemplateOperator mockRestTemplateOperator = mock(RestTemplateOperator.class);
         TaskSampleRetriever.getInstance().start(mockRestTemplateOperator);
 
         // test onStart and onStop if null handler
@@ -62,13 +68,13 @@ void testStartAndStop() {
 
         // test onStop if handler not empty
         ReflectionTestUtils.setField(observableAspectTask, "tableSampleHandlers", new HashMap<String, TableSampleHandler>() {{
-            put("test", Mockito.mock(TableSampleHandler.class));
+            put("test", mock(TableSampleHandler.class));
         }});
         ReflectionTestUtils.setField(observableAspectTask, "dataNodeSampleHandlers", new HashMap<String, DataNodeSampleHandler>() {{
-            put("test", Mockito.mock(DataNodeSampleHandler.class));
+            put("test", mock(DataNodeSampleHandler.class));
         }});
         ReflectionTestUtils.setField(observableAspectTask, "processorNodeSampleHandlers", new HashMap<String, ProcessorNodeSampleHandler>() {{
-            put("test", Mockito.mock(ProcessorNodeSampleHandler.class));
+            put("test", mock(ProcessorNodeSampleHandler.class));
         }});
         observableAspectTask.onStop(new TaskStopAspect().task(taskDto));
     }
@@ -116,7 +122,7 @@ void testCloseCompletableFuture1() {
                     interruptedException.printStackTrace();
                 }
             });
-            observableAspectTask.writeRecordFuture.thenRunAsync(() -> {
+            observableAspectTask.writeRecordFuture.runAsync(() -> {
                 try {
                     Thread.sleep(3000);
                 } catch (InterruptedException interruptedException) {
@@ -134,7 +140,6 @@ void testCloseCompletableFuture1() {
             Assertions.assertTrue(observableAspectTask.batchProcessFuture.isDone());
             Assertions.assertTrue(observableAspectTask.streamProcessFuture.isDone());
             Assertions.assertTrue(observableAspectTask.streamReadFuture.isDone());
-            Assertions.assertTrue(observableAspectTask.writeRecordFuture.isDone());
         }
     }
 
@@ -169,4 +174,62 @@ void testInterceptAspectTest0() {
             Assertions.assertNull(aspect);
         }
     }
+
+    @Nested
+    class handleDropTableFunc {
+        @Test
+        void testHandleDropTableFuncInitWithTrue() {
+            DropTableFuncAspect aspect = new DropTableFuncAspect();
+            aspect.setInit(true);
+            TaskSampleHandler taskSampleHandler =mock(TaskSampleHandler.class);
+            observableAspectTask.handleDropTableFunc(aspect);
+            verify(taskSampleHandler,times(0)).handleDdlEnd();
+        }
+        @Test
+        void testHandleDropTableFuncInitWithFalse() {
+            DropTableFuncAspect aspect = new DropTableFuncAspect();
+            aspect.setInit(false);
+            aspect.setState(20);
+            DataProcessorContext dataProcessorContext = mock(DataProcessorContext.class);
+            aspect.setDataProcessorContext(dataProcessorContext);
+            Node node = mock(Node.class);
+            node.setId("id");
+            when(dataProcessorContext.getNode()).thenReturn(node);
+            TaskSampleHandler taskSampleHandler =mock(TaskSampleHandler.class);
+            ReflectionTestUtils.setField(observableAspectTask,"taskSampleHandler",taskSampleHandler);
+            ReflectionTestUtils.setField(observableAspectTask,"dataNodeSampleHandlers",new HashMap());
+
+            observableAspectTask.handleDropTableFunc(aspect);
+            verify(taskSampleHandler,times(1)).handleDdlEnd();
+        }
+    }
+
+    @Nested
+    class handleCreateTableFunc {
+        @Test
+        void testHandleCreateTableFuncInitWithTrue() {
+            DropTableFuncAspect aspect = new DropTableFuncAspect();
+            aspect.setInit(true);
+            TaskSampleHandler taskSampleHandler =mock(TaskSampleHandler.class);
+            observableAspectTask.handleDropTableFunc(aspect);
+            verify(taskSampleHandler,times(0)).handleDdlEnd();
+        }
+        @Test
+        void testHandleCreateTableFuncInitWithFalse() {
+            DropTableFuncAspect aspect = new DropTableFuncAspect();
+            aspect.setInit(false);
+            aspect.setState(20);
+            DataProcessorContext dataProcessorContext = mock(DataProcessorContext.class);
+            aspect.setDataProcessorContext(dataProcessorContext);
+            Node node = mock(Node.class);
+            node.setId("id");
+            when(dataProcessorContext.getNode()).thenReturn(node);
+            TaskSampleHandler taskSampleHandler =mock(TaskSampleHandler.class);
+            ReflectionTestUtils.setField(observableAspectTask,"taskSampleHandler",taskSampleHandler);
+            ReflectionTestUtils.setField(observableAspectTask,"dataNodeSampleHandlers",new HashMap());
+
+            observableAspectTask.handleDropTableFunc(aspect);
+            verify(taskSampleHandler,times(1)).handleDdlEnd();
+        }
+    }
 }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureExTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureExTest.java
new file mode 100644
index 0000000000..47444915e4
--- /dev/null
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureExTest.java
@@ -0,0 +1,110 @@
+package io.tapdata.observable.metric.util;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * @author samuel
+ * @Description
+ * @create 2024-06-28 14:13
+ **/
+class TapCompletableFutureExTest {
+	@Test
+	@DisplayName("Method create test")
+	void testCreate() {
+		TapCompletableFutureEx tapCompletableFutureEx = TapCompletableFutureEx.create(10000, 1000);
+
+		assertEquals(10000, tapCompletableFutureEx.getQueueSize());
+		assertEquals(1000, tapCompletableFutureEx.getJoinWatermark());
+		Object running = ReflectionTestUtils.getField(tapCompletableFutureEx, "running");
+		assertNotNull(running);
+		assertFalse(((AtomicBoolean) running).get());
+		Object completableFutureQueue = ReflectionTestUtils.getField(tapCompletableFutureEx, "completableFutureQueue");
+		assertNotNull(completableFutureQueue);
+		assertEquals(10000, ((LinkedBlockingQueue<?>) completableFutureQueue).remainingCapacity());
+		Object joinThreadPool = ReflectionTestUtils.getField(tapCompletableFutureEx, "joinThreadPool");
+		assertNotNull(joinThreadPool);
+		assertInstanceOf(ThreadPoolExecutor.class, joinThreadPool);
+		assertEquals(1, ((ThreadPoolExecutor) joinThreadPool).getCorePoolSize());
+		assertEquals(1, ((ThreadPoolExecutor) joinThreadPool).getMaximumPoolSize());
+
+		tapCompletableFutureEx = TapCompletableFutureEx.create(0, 0);
+		assertEquals(TapCompletableFutureEx.DEFAULT_QUEUE_SIZE, tapCompletableFutureEx.getQueueSize());
+		assertEquals(TapCompletableFutureEx.DEFAULT_JOIN_WATERMARK, tapCompletableFutureEx.getJoinWatermark());
+	}
+
+	@Test
+	@DisplayName("Method start test")
+	void testStart() {
+		TapCompletableFutureEx tapCompletableFutureEx = TapCompletableFutureEx.create(10000, 1000).start();
+
+		Object running = ReflectionTestUtils.getField(tapCompletableFutureEx, "running");
+		assertNotNull(running);
+		assertTrue(((AtomicBoolean) running).get());
+		Object joinThreadPool = ReflectionTestUtils.getField(tapCompletableFutureEx, "joinThreadPool");
+		assertNotNull(joinThreadPool);
+		assertEquals(1, ((ThreadPoolExecutor) joinThreadPool).getActiveCount());
+	}
+
+	@Test
+	@DisplayName("Method runAsync test, single thread")
+	void testRunAsyncSingleThread() {
+		TapCompletableFutureEx tapCompletableFutureEx = TapCompletableFutureEx.create(1000, 1000).start();
+		AtomicBoolean run = new AtomicBoolean(false);
+		tapCompletableFutureEx.runAsync(() -> {
+			try {
+				TimeUnit.SECONDS.sleep(1L);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			run.set(true);
+		});
+		tapCompletableFutureEx.stop(1L, TimeUnit.MINUTES);
+		assertTrue(run.get());
+	}
+
+	@Test
+	@DisplayName("Method runAsync test, multi thread")
+	void testRunAsyncMultiThread() {
+		int threadNum = 4, threadBatch = 1000000;
+		long start = System.currentTimeMillis();
+		List<CompletableFuture<Void>> completableFutures = new ArrayList<>();
+		TapCompletableFutureEx tapCompletableFutureEx = TapCompletableFutureEx.create(Integer.MAX_VALUE, threadBatch).start();
+		AtomicInteger counter = new AtomicInteger();
+		for (int i = 0; i < threadNum; i++) {
+			completableFutures.add(
+					CompletableFuture.runAsync(() -> {
+						for (int j = 0; j < threadBatch; j++) {
+							assertDoesNotThrow(() -> tapCompletableFutureEx.runAsync(counter::incrementAndGet));
+						}
+					})
+			);
+		}
+
+		assertDoesNotThrow(() -> CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join());
+		assertDoesNotThrow(() -> tapCompletableFutureEx.stop(5L, TimeUnit.MINUTES));
+		assertEquals(threadNum * threadBatch, counter.get());
+		long cost = System.currentTimeMillis() - start;
+		BigDecimal qps;
+		if ((cost / 1000) == 0) {
+			qps = BigDecimal.valueOf(threadBatch * threadNum);
+		} else {
+			qps = BigDecimal.valueOf(threadNum * threadBatch).divide(BigDecimal.valueOf(cost / 1000), 2, RoundingMode.HALF_UP);
+		}
+		System.out.println("cost: " + cost + ", qps: " + qps);
+	}
+}
\ No newline at end of file
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskExTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskExTest.java
new file mode 100644
index 0000000000..c31f9a37c0
--- /dev/null
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/util/TapCompletableFutureTaskExTest.java
@@ -0,0 +1,59 @@
+package io.tapdata.observable.metric.util;
+
+import com.tapdata.tm.commons.task.dto.TaskDto;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * @author samuel
+ * @Description
+ * @create 2024-06-28 14:12
+ **/
+class TapCompletableFutureTaskExTest {
+
+	private TaskDto taskDto;
+
+	@BeforeEach
+	void setUp() {
+		taskDto = new TaskDto();
+		taskDto.setId(new ObjectId());
+		taskDto.setName("test");
+	}
+
+	@Test
+	@DisplayName("Method create test")
+	void testCreate() {
+		TapCompletableFutureTaskEx tapCompletableFutureTaskEx = TapCompletableFutureTaskEx.create(1000, 1000, taskDto);
+		assertSame(taskDto, tapCompletableFutureTaskEx.getTaskDto());
+
+		assertDoesNotThrow(() -> TapCompletableFutureTaskEx.create(1000, 1000, null));
+		taskDto.setId(null);
+		assertDoesNotThrow(() -> TapCompletableFutureTaskEx.create(1000, 1000, taskDto));
+		taskDto.setName(null);
+		assertDoesNotThrow(() -> TapCompletableFutureTaskEx.create(1000, 100, taskDto));
+	}
+
+	@Test
+	@DisplayName("Test runAsync and stop with single thread")
+	void testRunAsyncAndStopSingleThread() {
+		TapCompletableFutureEx tapCompletableFutureTaskEx = TapCompletableFutureTaskEx.create(1000, 1000, taskDto).start();
+		AtomicInteger counter = new AtomicInteger();
+		tapCompletableFutureTaskEx.runAsync(() -> {
+			try {
+				TimeUnit.SECONDS.sleep(1L);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			counter.incrementAndGet();
+		});
+		tapCompletableFutureTaskEx.stop(5L, TimeUnit.SECONDS);
+		assertEquals(1, counter.get());
+	}
+}
\ No newline at end of file
diff --git a/manager/tm-api/src/test/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtilsTest.java b/manager/tm-api/src/test/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtilsTest.java
new file mode 100644
index 0000000000..b76f588358
--- /dev/null
+++ b/manager/tm-api/src/test/java/com/tapdata/tm/inspect/recovery/AutoRecoveryUtilsTest.java
@@ -0,0 +1,52 @@
+package com.tapdata.tm.inspect.recovery;
+
+import com.tapdata.tm.inspect.constant.InspectMethod;
+import com.tapdata.tm.inspect.constant.InspectStatusEnum;
+import com.tapdata.tm.inspect.dto.InspectDto;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+/**
+ * 自动修复工具类测试
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/6/14 14:15 Create
+ */
+class AutoRecoveryUtilsTest {
+    @Nested
+    class CheckCanRecoveryTest {
+        @Test
+        void testTrue() {
+            InspectDto inspectDto = new InspectDto();
+            inspectDto.setInspectMethod(InspectMethod.FIELD.getValue());
+            inspectDto.setStatus(InspectStatusEnum.DONE.getValue());
+            inspectDto.setResult("failed");
+            inspectDto.setFlowId("test-flow-id");
+
+            List<String> errors = AutoRecoveryUtils.checkCanRecovery(inspectDto);
+            Assertions.assertTrue(errors.isEmpty());
+
+            inspectDto.setInspectMethod(InspectMethod.JOINTFIELD.getValue());
+            errors = AutoRecoveryUtils.checkCanRecovery(inspectDto);
+            Assertions.assertTrue(errors.isEmpty());
+        }
+
+        @Test
+        void testFalse() {
+            InspectDto inspectDto = new InspectDto();
+            inspectDto.setInspectMethod(InspectMethod.CDC_COUNT.getValue());
+            inspectDto.setStatus(InspectStatusEnum.ERROR.getValue());
+            inspectDto.setResult("error");
+
+            List<String> errors = AutoRecoveryUtils.checkCanRecovery(inspectDto);
+            Assertions.assertFalse(errors.isEmpty());
+
+            inspectDto.setFlowId("");
+            errors = AutoRecoveryUtils.checkCanRecovery(inspectDto);
+            Assertions.assertFalse(errors.isEmpty());
+        }
+    }
+}
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/dag/DAGTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/dag/DAGTest.java
index b55a52186c..5be4db9d12 100644
--- a/manager/tm-common/src/test/java/com/tapdata/tm/commons/dag/DAGTest.java
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/dag/DAGTest.java
@@ -1,24 +1,20 @@
 package com.tapdata.tm.commons.dag;
 
 import com.tapdata.tm.commons.dag.nodes.DataParentNode;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
 import com.tapdata.tm.commons.dag.process.JsProcessorNode;
+import com.tapdata.tm.commons.dag.process.TableRenameProcessNode;
+import com.tapdata.tm.commons.dag.vo.TableRenameTableInfo;
 import com.tapdata.tm.commons.task.dto.Message;
 import org.bson.types.ObjectId;
 import org.junit.jupiter.api.*;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
 import static org.mockito.internal.verification.VerificationModeFactory.times;
 
 public class DAGTest {
@@ -40,6 +36,69 @@ void init() {
             when(dag.getTaskDtoIsomorphism(anyList())).thenCallRealMethod();
             when(dag.getTaskDtoIsomorphism(null)).thenCallRealMethod();
         }
+        @Nested
+        class getConvertTableNameMapTest{
+            TableRenameProcessNode tableRenameProcessNode;
+            @BeforeEach
+            void setUp(){
+                tableRenameProcessNode = mock(TableRenameProcessNode.class);
+            }
+            @DisplayName("test get convert table name map")
+            @Test
+            void test1(){
+                String tableName="beforeTableName";
+                ArrayList<String> tableNames=new ArrayList<>();
+                tableNames.add(tableName);
+                Map<String, TableRenameTableInfo> tableRenameTableInfoMap = new HashMap<>();
+                when(tableRenameProcessNode.originalMap()).thenReturn(tableRenameTableInfoMap);
+                when(tableRenameProcessNode.convertTableName(anyMap(),anyString(),eq(false))).thenReturn("convertNewTableName");
+                Map<String, String> convertTableNameMap = DAG.getConvertTableNameMap(tableRenameProcessNode, tableNames);
+                assertEquals("convertNewTableName",convertTableNameMap.get("beforeTableName"));
+            }
+        }
+        @Nested
+        class GenerateTableNameRelationTest{
+            DatabaseNode sourceNode;
+            DatabaseNode targetNode;
+
+            @BeforeEach
+            void setUp(){
+                sourceNode=mock(DatabaseNode.class);
+                targetNode=mock(DatabaseNode.class);
+            }
+            @DisplayName("test generateTableNameRelation when hava rename node")
+            @Test
+            void test1(){
+                String tableName="beforeTableName";
+                ArrayList<String> tableNames=new ArrayList<>();
+                tableNames.add(tableName);
+                TableRenameProcessNode tableRenameProcessNode = mock(TableRenameProcessNode.class);
+                Map<String, TableRenameTableInfo> tableRenameTableInfoMap = new HashMap<>();
+                when(tableRenameProcessNode.originalMap()).thenReturn(tableRenameTableInfoMap);
+                when(tableRenameProcessNode.convertTableName(anyMap(),anyString(),eq(false))).thenReturn("convertNewTableName");
+                LinkedList<Node> nodes=new LinkedList<>();
+                nodes.add(sourceNode);
+                nodes.add(targetNode);
+                nodes.add(tableRenameProcessNode);
+                LinkedHashMap<String, String> tableNameRelation = DAG.generateTableNameRelation(nodes, tableNames);
+                assertEquals("convertNewTableName",tableNameRelation.get("beforeTableName"));
+                assertEquals("convertNewTableName",tableNames.get(0));
+            }
+            @DisplayName("test generateTableNameRelation when not hava rename node")
+            @Test
+            void test2(){
+                String tableName="beforeTableName";
+                ArrayList<String> tableNames=new ArrayList<>();
+                tableNames.add(tableName);
+                LinkedList<Node> nodes=new LinkedList<>();
+                nodes.add(sourceNode);
+                nodes.add(targetNode);
+                LinkedHashMap<String, String> tableNameRelation = DAG.generateTableNameRelation(nodes, tableNames);
+                assertEquals("beforeTableName",tableNameRelation.get("beforeTableName"));
+                assertEquals("beforeTableName",tableNames.get(0));
+            }
+
+        }
 
         @Test
         void testGetTaskDtoIsomorphismNormal() {
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/utils/ConnHeartbeatUtilsTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/utils/ConnHeartbeatUtilsTest.java
new file mode 100644
index 0000000000..628c510696
--- /dev/null
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/utils/ConnHeartbeatUtilsTest.java
@@ -0,0 +1,63 @@
+package com.tapdata.tm.utils;
+
+import com.tapdata.tm.commons.task.dto.ParentTaskDto;
+import com.tapdata.tm.commons.task.dto.TaskDto;
+import com.tapdata.tm.commons.util.ConnHeartbeatUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+public class ConnHeartbeatUtilsTest {
+    @Nested
+    class CheckTask{
+        @Test
+        void testCacheTask(){
+            Assertions.assertTrue(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_MEM_CACHE));
+        }
+
+        @Test
+        void testCacheTask_initial_sync(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC, TaskDto.SYNC_TYPE_MEM_CACHE));
+        }
+
+        @Test
+        void testMigrateTask(){
+            Assertions.assertTrue(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_MIGRATE));
+        }
+
+        @Test
+        void testMigrateTask_initial_sync(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC, TaskDto.SYNC_TYPE_MIGRATE));
+        }
+
+        @Test
+        void testSyncTask(){
+            Assertions.assertTrue(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_SYNC));
+        }
+
+        @Test
+        void testSyncTask_initial_sync(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC, TaskDto.SYNC_TYPE_SYNC));
+        }
+
+        @Test
+        void testLogCollectorTask(){
+            Assertions.assertTrue(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_LOG_COLLECTOR));
+        }
+
+        @Test
+        void testLogCollectorTask_initial_sync(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC, TaskDto.SYNC_TYPE_LOG_COLLECTOR));
+        }
+
+        @Test
+        void testConnHeartbeatTask(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_CONN_HEARTBEAT));
+        }
+
+        @Test
+        void testTestRunTask(){
+            Assertions.assertFalse(ConnHeartbeatUtils.checkTask(ParentTaskDto.TYPE_INITIAL_SYNC_CDC, TaskDto.SYNC_TYPE_TEST_RUN));
+        }
+    }
+}
diff --git a/manager/tm-oss/src/test/java/com/tapdata/tm/inspect/service/InspectServiceImplTest.java b/manager/tm-oss/src/test/java/com/tapdata/tm/inspect/service/InspectServiceImplTest.java
index 27c3528c8d..bf522e7897 100644
--- a/manager/tm-oss/src/test/java/com/tapdata/tm/inspect/service/InspectServiceImplTest.java
+++ b/manager/tm-oss/src/test/java/com/tapdata/tm/inspect/service/InspectServiceImplTest.java
@@ -363,31 +363,5 @@ void findAllByIds() {
         }
     }
 
-    @Test
-    void fieldHandler() {
-        doCallRealMethod().when(impl).fieldHandler(anyList(), any(UserDetail.class));
-        try(MockedStatic<MessageUtil> messageUtilMockedStatic = Mockito.mockStatic(MessageUtil.class)) {
-            messageUtilMockedStatic.when(() -> MessageUtil.getMessage(anyString())).thenReturn("error");
-            Assertions.assertThrows(BizException.class, () -> {
-                try {
-                    impl.fieldHandler(mock(List.class), mock(UserDetail.class));
-                } catch (BizException e) {
-                    Assertions.assertEquals(ConstVariable.TA_OSS_NON_SUPPORT_FUNCTION_EXCEPTION, e.getErrorCode());
-                    throw e;
-                }
-            });
-        }
-    }
-
-    @Nested
-    class FieldHandler{
-        @Test
-        void test(){
-            try(MockedStatic<MessageUtil> messageUtilMockedStatic = Mockito.mockStatic(MessageUtil.class)){
-                messageUtilMockedStatic.when(()->MessageUtil.getMessage(anyString())).thenReturn("error");
-                InspectServiceImpl inspectService = new InspectServiceImpl(mock(InspectRepository.class));
-                Assertions.assertThrows(BizException.class,()->inspectService.fieldHandler(null,null));
-            }
-        }
-    }
+    
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/discovery/service/DiscoveryServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/discovery/service/DiscoveryServiceImplTest.java
new file mode 100644
index 0000000000..f505e54b48
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/discovery/service/DiscoveryServiceImplTest.java
@@ -0,0 +1,318 @@
+package com.tapdata.tm.discovery.service;
+
+import com.tapdata.tm.base.exception.BizException;
+import com.tapdata.tm.config.security.UserDetail;
+import com.tapdata.tm.livedataplatform.dto.LiveDataPlatformDto;
+import com.tapdata.tm.livedataplatform.service.LiveDataPlatformService;
+import com.tapdata.tm.metadatadefinition.dto.MetadataDefinitionDto;
+import com.tapdata.tm.metadatadefinition.vo.MetaDataDefinitionVo;
+import org.bson.Document;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author samuel
+ * @Description
+ * @create 2024-06-20 17:00
+ **/
+class DiscoveryServiceImplTest {
+
+	private DiscoveryServiceImpl discoveryService;
+
+	@BeforeEach
+	void setUp() {
+		discoveryService = new DiscoveryServiceImpl();
+	}
+
+	@Nested
+	@DisplayName("Method isMDMRoot test")
+	class isMDMRootTest {
+		@Test
+		@DisplayName("test main process")
+		void test1() {
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			metadataDefinitionDto.setParent_id(null);
+			metadataDefinitionDto.setItemType(new ArrayList<String>() {{
+				add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			}});
+			assertTrue(discoveryService.isMDMRoot(metadataDefinitionDto));
+
+			metadataDefinitionDto.setParent_id("1");
+			assertFalse(discoveryService.isMDMRoot(metadataDefinitionDto));
+
+			metadataDefinitionDto.setParent_id(null);
+			metadataDefinitionDto.setItemType(new ArrayList<String>() {{
+				add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			}});
+			assertFalse(discoveryService.isMDMRoot(metadataDefinitionDto));
+		}
+
+		@Test
+		@DisplayName("test input null")
+		void test2() {
+			assertFalse(discoveryService.isMDMRoot(null));
+		}
+	}
+
+	@Nested
+	@DisplayName("Method addMetadataCriteriaListTags test")
+	class addMetadataCriteriaListTagsTest {
+		@Test
+		@DisplayName("test is mdm root")
+		void test1() {
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			metadataDefinitionDto.setParent_id(null);
+			metadataDefinitionDto.setItemType(new ArrayList<String>() {{
+				add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			}});
+			List<String> tagIds = new ArrayList<>();
+			tagIds.add("1");
+			Criteria criteria = new Criteria();
+			discoveryService.addMetadataCriteriaListTags(metadataDefinitionDto, tagIds, criteria);
+
+			Document criteriaObject = criteria.getCriteriaObject();
+			assertEquals("{\"$or\": [{\"listtags.id\": {\"$exists\": false}}, {\"listtags.id\": {\"$in\": [\"1\"]}}]}", criteriaObject.toJson());
+		}
+
+		@Test
+		@DisplayName("test is not mdm root")
+		void test2() {
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			metadataDefinitionDto.setParent_id("parent1");
+			metadataDefinitionDto.setItemType(new ArrayList<String>() {{
+				add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			}});
+			List<String> tagIds = new ArrayList<>();
+			tagIds.add("tag1");
+			Criteria criteria = new Criteria();
+			discoveryService.addMetadataCriteriaListTags(metadataDefinitionDto, tagIds, criteria);
+
+			assertEquals("{\"listtags.id\": {\"$in\": [\"tag1\"]}}", criteria.getCriteriaObject().toJson());
+		}
+
+		@Test
+		@DisplayName("test tagIds is null")
+		void test3() {
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			metadataDefinitionDto.setParent_id(null);
+			metadataDefinitionDto.setItemType(new ArrayList<String>() {{
+				add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			}});
+			Criteria criteria = new Criteria();
+			discoveryService.addMetadataCriteriaListTags(metadataDefinitionDto, null, criteria);
+
+			assertEquals("{\"$or\": [{\"listtags.id\": {\"$exists\": false}}]}", criteria.getCriteriaObject().toJson());
+
+			metadataDefinitionDto.setParent_id("parent1");
+			criteria = new Criteria();
+			discoveryService.addMetadataCriteriaListTags(metadataDefinitionDto, null, criteria);
+
+			assertTrue(criteria.getCriteriaObject().isEmpty());
+		}
+	}
+
+	@Nested
+	@DisplayName("Method addMetadataCriteriaMDMConnId test")
+	class addMetadataCriteriaMDMConnIdTest {
+		@Test
+		@DisplayName("test main process")
+		void test1() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"fdm_conn_id\", \"mdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+		}
+
+		@Test
+		@DisplayName("test ldp not exists")
+		void test2() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(null);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			metadataDefinitionDto.setItemType(itemType);
+			BizException bizException = assertThrows(BizException.class, () -> discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto));
+
+			assertEquals("Ldp.not.exists", bizException.getErrorCode());
+		}
+
+		@Test
+		@DisplayName("test metadataDefinitionDto is null")
+		void test3() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+
+			assertDoesNotThrow(()-> discoveryService.addMetadataCriteriaMDMConnId(user, criteria, null));
+
+			assertTrue(criteria.getCriteriaObject().isEmpty());
+		}
+
+		@Test
+		@DisplayName("test itemType only have fdm")
+		void test4() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"fdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+		}
+
+		@Test
+		@DisplayName("test itemType only have mdm")
+		void test5() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"mdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+		}
+
+		@Test
+		@DisplayName("test itemType don't have fdm and mdm")
+		void test6() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.ITEM_TYPE_APP);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertTrue(criteria.getCriteriaObject().isEmpty());
+		}
+
+		@Test
+		@DisplayName("test itemType is null")
+		void test7() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			metadataDefinitionDto.setItemType(null);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertTrue(criteria.getCriteriaObject().isEmpty());
+		}
+
+		@Test
+		@DisplayName("test fdm connection id is blank")
+		void test8() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId(null);
+			liveDataPlatformDto.setMdmStorageConnectionId("mdm_conn_id");
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"mdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+
+			criteria = new Criteria();
+			liveDataPlatformDto.setFdmStorageConnectionId("");
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"mdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+		}
+
+		@Test
+		@DisplayName("test mdm connection id is blank")
+		void test9() {
+			UserDetail user = mock(UserDetail.class);
+			Criteria criteria = new Criteria();
+			LiveDataPlatformService liveDataPlatformService = mock(LiveDataPlatformService.class);
+			ReflectionTestUtils.setField(discoveryService, "liveDataPlatformService", liveDataPlatformService);
+			LiveDataPlatformDto liveDataPlatformDto = new LiveDataPlatformDto();
+			liveDataPlatformDto.setFdmStorageConnectionId("fdm_conn_id");
+			liveDataPlatformDto.setMdmStorageConnectionId(null);
+			when(liveDataPlatformService.findOne(any(Query.class))).thenReturn(liveDataPlatformDto);
+			MetadataDefinitionDto metadataDefinitionDto = new MetadataDefinitionDto();
+			List<String> itemType = new ArrayList<>();
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_FDM);
+			itemType.add(MetadataDefinitionDto.LDP_ITEM_MDM);
+			metadataDefinitionDto.setItemType(itemType);
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"fdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+
+			criteria = new Criteria();
+			liveDataPlatformDto.setMdmStorageConnectionId("");
+			discoveryService.addMetadataCriteriaMDMConnId(user, criteria, metadataDefinitionDto);
+
+			assertEquals("{\"source._id\": {\"$in\": [\"fdm_conn_id\"]}}", criteria.getCriteriaObject().toJson());
+		}
+	}
+}
\ No newline at end of file