Repository: tapdata/tapdata (language: python)

Issue:
Tap 5257 dev



Discussion / hints:

Fix patch:
diff --git a/RELEASE b/RELEASE
index dadc8cfd0e..0d49d2485e 100644
--- a/RELEASE
+++ b/RELEASE
@@ -1,2 +1,2 @@
-LST: v3.16.0
-DEV: v3.17.0-dev
\ No newline at end of file
+LST: v3.17.0
+DEV: v3.18.0-dev
\ No newline at end of file
diff --git a/iengine/api/src/main/java/io/tapdata/loglistener/TapLogger.java b/iengine/api/src/main/java/io/tapdata/loglistener/TapLogger.java
index dddb6c3175..54fe8e83de 100644
--- a/iengine/api/src/main/java/io/tapdata/loglistener/TapLogger.java
+++ b/iengine/api/src/main/java/io/tapdata/loglistener/TapLogger.java
@@ -7,6 +7,7 @@ public interface TapLogger {
     void warn(String message, Object... params);
     void error(String message, Object... params);
     void debug(String message, Object... params);
+    void trace(String message, Object... params);
     boolean isDebugEnabled();
 
 
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/Schedule/ConnectorManager.java b/iengine/iengine-app/src/main/java/io/tapdata/Schedule/ConnectorManager.java
index 3d6a5a728e..987a56a650 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/Schedule/ConnectorManager.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/Schedule/ConnectorManager.java
@@ -619,7 +619,7 @@ public SchemaProxy initSchemaProxy() {
 		return SchemaProxy.SchemaProxyInstance.INSTANCE.getInstance(clientMongoOperator);
 	}
 
-	@Scheduled(fixedDelay = 2000L)
+	@Scheduled(fixedDelay = 1000L * 60 * 60)
 	public void refreshToken() throws InterruptedException {
 		Thread.currentThread().setName(String.format(ConnectorConstant.REFREASH_TOKEN_THREAD, CONNECTOR, instanceNo.substring(instanceNo.length() - 6)));
 		User user = (User) configCenter.getConfig(ConfigurationCenter.USER_INFO);
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/error/TaskMergeProcessorExCode_16.java b/iengine/iengine-app/src/main/java/io/tapdata/error/TaskMergeProcessorExCode_16.java
index f06415c1d8..887afdb435 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/error/TaskMergeProcessorExCode_16.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/error/TaskMergeProcessorExCode_16.java
@@ -12,64 +12,115 @@
 public interface TaskMergeProcessorExCode_16 {
 	@TapExCode
 	String UNKNOWN_ERROR = "16001";
-
+	//静态信息可以剪短 joinkey ,node name, table name
 	@TapExCode(
 			describe = "Failed to look up data from the cache based on the federated key when looking up the data in the subtable",
-			describeCN = "反查子表数据时，根据联合键从缓存中查找数据失败"
+			describeCN = "反查子表数据时，根据联合键从缓存中查找数据失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Main table node name :{}, reverse lookup sub-table node name :{}, union key value :{}, encoded union key value :{}, configured external storage database type :{}, external storage name :{}, storage structure table name :{} ",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"主表节点名称:{}，反查子表的节点名称:{}，联合键值:{}，编码后的联合键值:{}，配置的外存数据库类型为:{}，外存名称为:{}，存储结构表名：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String LOOK_UP_FIND_BY_JOIN_KEY_FAILED = "16002";
 
 	@TapExCode(
-			describe = "Merge table node lookup failed, from node id list is empty",
-			describeCN = "合并表数据反查失败，from节点id列表为空"
+			describe = "When the subtable data is unlooked up, the list of node ids passed by the acquisition event is empty, so the configuration of the master-slave merge cannot be obtained",
+			describeCN = "反查子表数据时，获取事件经过的节点ID列表为空,导致无法获取主从合并的配置"
 	)
 	String LOOK_UP_MISSING_FROM_NODE_ID = "16003";
 
 	@TapExCode(
-			describe = "Cannot found pre node",
-			describeCN = "无法找到上个节点"
+			describe = "The previous node of the master-slave merged node cannot be obtained by the node ID",
+			describeCN = "无法通过节点ID获取主从合并节点的上一个节点"
 	)
 	String CANNOT_FOUND_PRE_NODE = "16004";
 
 	@TapExCode(
-			describe = "Wrong node type",
-			describeCN = "节点类型错误"
+			describe = "When initializing the master-slave merge node, the unexpected node type is detected",
+			describeCN = "初始化主从合并节点时，检测出非预期节点类型",
+			dynamicDescription = "Expect MergeTableNode, but got: {}",
+			dynamicDescriptionCN = "预期是MergeTableNode，但是检测出{}"
 	)
 	String WRONG_NODE_TYPE = "16005";
 
 	@TapExCode(
 			describe = "The merged source table does not have a primary key or a unique index, the data cannot be cached, and the merge cannot be performed normally",
-			describeCN = "合并的源表没有主键或唯一索引，无法对数据缓存，合并无法正常进行"
+			describeCN = "合并的源表没有主键或唯一索引，无法对数据缓存，合并无法正常进行",
+			dynamicDescription = "The absence of a primary key or unique index in the source table makes it impossible to uniquely identify data when caching\n" +
+					"Table name: {} Node name: {}",
+			dynamicDescriptionCN = "源表中缺少主键或唯一索引，导致缓存数据时无法对数据进行唯一标识\n" +
+					"表名：{}，节点名：{}，合并操作：{}",
+			solution = "1. Add a primary key or unique index to the table in the database of the source table, based on the table name suggested\n" +
+					"2. Add the enhanced JS node after the node name indicating the missing primary key. In the enhanced JS basic Settings, click Use Model Declaration and add the TapModelDeclare.setPk(tapTable, 'fieldName') script to the script to set the logical primary key for the source table. Restart tasks",
+			solutionCN = "1. 根据提示的表名，在源表的数据库中为表添加主键或唯一索引\n" +
+					"2. 根据提示缺少主键的的节点名，在对应的节点后面添加增强JS节点。在增强JS的基础设置中点击使用模型声明，并在脚本中添加\n" +
+					"<pre><code>TapModelDeclare.setPk(tapTable, 'fieldName')" +
+					"</code></pre>\n" +
+					"为源表设置逻辑主键后。重新启动任务"
 	)
 	String TAP_MERGE_TABLE_NO_PRIMARY_KEY = "16006";
 
 	@TapExCode(
-			describe = "The mode of merging into the array lacks the associated key of the array, please set it in the interface",
-			describeCN = "合并进数组的模式，缺少数组关联键，请在界面设置"
+			describe = "When the data write mode of the child table is \"update into embedded data\", the embedded array matching condition is not filled",
+			describeCN = "当子表的数据写入模式为“更新进内嵌数据”时,内嵌数组匹配条件没有填写",
+			dynamicDescription = "The master-slave configuration of the master-slave merge node lacks a condition for the embedded array to match, so the child table data updates cannot be written to the embedded array\n" +
+					"Table name: {} Node name: {}",
+			dynamicDescriptionCN = "主从合并节点的主从配置缺少内嵌数组匹配的条件，导致子表数据更新无法写入到内嵌数组里" +
+					"表名：{}，节点名：{}",
+			solution = "1. According to the table name and node name indicated in the error message, find the corresponding master-slave configuration in the master-slave merge and set the matching condition of the embedded array to restart the task",
+			solutionCN = "1. 根据报错信息中提示的表名与节点名，找到主从合并中对应的主从配置并设置内嵌数组匹配条件后重新启动任务"
 	)
 	String TAP_MERGE_TABLE_NO_ARRAY_KEY = "16007";
 
 	@TapExCode(
 			describe = "Updating or writing to the cache, finding data from the cache based on the federated key failed",
-			describeCN = "更新或写入缓存，根据联合键从缓存中查找数据失败"
+			describeCN = "更新或写入缓存时，根据联合键从缓存中查找原有数据失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"node name: {}, Table name: {},  association key: {}, encoded association key: {}, primary key or unique index key: {}, encoded primary key or unique index key: {}, configured external storage database type: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，关联键值：{}，编码后的关联键值：{}，主键或唯一键值：{}，编码后的主键或唯一键值：{}，更新或写入的数据：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String UPSERT_CACHE_FIND_BY_JOIN_KEY_FAILED = "16008";
+	//表名、节点、字段名原值、拼起来后的joinkey,外存类型，外存的表名
 
 	@TapExCode(
 			describe = "Update or write cache failed",
-			describeCN = "更新或写入缓存失败"
+			describeCN = "更新或写入缓存失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Node name: {}, table name: {}, association key: {}, encoded association key: {}, primary or unique key: {}, encoded primary or unique key: {}, updated or written data: {}, configured external storage database type: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，关联键值：{}，编码后的关联键值：{}，主键或唯一键值：{}，编码后的主键或唯一键值：{}，更新或写入的数据：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String UPSERT_CACHE_FAILED = "16009";
 
+
 	@TapExCode(
 			describe = "Deleting cache, finding data from the cache based on the federated key failed",
-			describeCN = "删除缓存，根据联合键从缓存中查找数据失败"
+			describeCN = "删除缓存时，根据联合键从缓存中查找数据失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Node name: {}, table name: {}, association key value: {}, encoded association key value: {}, primary or unique key value: {}, encoded primary or unique key value: {}, configured external storage database type: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，关联键值：{}，编码后的关联键值：{}，主键或唯一键值：{}，编码后的主键或唯一键值：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String DELETE_CACHE_FIND_BY_JOIN_KEY_FAILED = "16010";
 
 	@TapExCode(
 			describe = "Delete cache failed",
-			describeCN = "删除缓存失败"
+			describeCN = "删除缓存失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"The external storage database type configured is :{}, the external storage name is :{}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，关联键值：{}，编码后的关联键值：{}，主键或唯一键值：{}，编码后的主键或唯一键值：{}，更新或写入的数据：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String DELETE_CACHE_FAILED = "16011";
 
@@ -111,25 +162,48 @@ public interface TaskMergeProcessorExCode_16 {
 
 	@TapExCode(
 			describe = "In the master-slave merge configuration, the source join key configuration is missing",
-			describeCN = "主从合并配置中，缺少源端关联键配置"
+			describeCN = "主从合并节点的配置中，缺少源端关联条件配置",
+			dynamicDescription = "Node name: {} The subsequent master-slave merge node is missing the associated key configuration",
+			dynamicDescriptionCN = "Node name: {} The subsequent master-slave merge node is missing the associated key configuration. mergeProperty: {}",
+			solution = "Please find the master-slave merge node according to the node name prompted, and configure the association condition in the corresponding table and fill in the association condition",
+			solutionCN = "请根据提示的节点名称找到主从合并节点，并在找到对应的表配置关联条件并填写关联条件"
+
 	)
 	String MISSING_SOURCE_JOIN_KEY_CONFIG = "16018";
 
 	@TapExCode(
-			describe = "The value for the join key does not exist in the data",
-			describeCN = "数据中不存在关联键的值"
+			describe = "When the data passes through the master-slave merge node, the value of the associated condition field set in the master-slave merge node is missing in the data",
+			describeCN = "数据经过主从合并节点时，数据中缺少主从合并节点中设置的关联条件字段的值，可能的原因是\n" +
+					"1. 在主从合并节点前、源节点后，存在处理节点，如JS节点。处理节点将数据中的关联字段值移除，导致错误。\n" +
+					"2. 主从合并节点的源节点的数据库为非结构化数据库，源节点某条数据中缺少关联字段的值，导致错误。",
+			dynamicDescription = "The data flowing through a node named \"{}\" is missing an associated key. Associated key: {}, data content: {}",
+			dynamicDescriptionCN = "经过节点名为“{}”的节点流出的数据中缺少关联键。关联键：{}，数据内容：{}",
+			solution = "1. If the source database of the master-slave merge node is an unstructured database such as MongoDB. The value of the associated field is missing from the data. You can add other processing nodes such as JS nodes after the source node and before the master-slave merging node, and add the script code to filter the missing associated key data, so as not to transmit to the master-slave merging node to ensure the normal operation of the task. \n"+
+					"2. If there are other processing nodes such as JS nodes before the master-slave merge node, check if there is code in the script to remove the associated key in the data or remove the processing node after the source node",
+			solutionCN = "1. 如果主从合并节点的源节点的数据库为非结构化数据库，如MongoDB。数据中缺少关联字段的值。可以在源节点的后面、主从合并节点前加入其他处理节点如JS节点，并在脚本中加入将缺少关联键数据过滤的脚本代码，不往主从合并节点传输，保证任务正常运行\n" +
+					"2. 如果主从合并节点前存在其他处理节点如JS节点，请检查脚本中是否存在将数据中的关联键移除的代码或将源节点后继的处理节点删除"
 	)
 	String JOIN_KEY_VALUE_NOT_EXISTS = "16019";
 
 	@TapExCode(
 			describe = "In the master-slave merge configuration, the target join key configuration is missing",
-			describeCN = "主从合并配置中，缺少目标关联键配置"
+			describeCN = "主从合并配置中，缺少目标关联条件配置",
+			dynamicDescription = "Node name: {} The subsequent master-slave merge node is missing the associated key configuration. mergeProperty: {}",
+			dynamicDescriptionCN = "节点名：{}后继的主从合并节点缺少关联键配置。合并属性：{}",
+			solution = "Please find the master-slave merge node according to the node name prompted, and find the corresponding table configuration association condition according to the node name in the master-slave merge node configuration",
+			solutionCN = "请根据提示的节点名称找到主从合并节点，并在主从合并节点配置中根据节点名称找到对应的表配置关联条件"
 	)
 	String MISSING_TARGET_JOIN_KEY_CONFIG = "16020";
 
 	@TapExCode(
-			describe = "The value of the primary key or unique index does not exist in the data",
-			describeCN = "数据中不存在主键或唯一索引的值"
+			describe = "Possible reasons for the absence of a primary key or unique index value in the data when it passes through a master-slave merge node\n" +
+					"Before the master-slave merge node and after the source node, there are processing nodes, such as JS nodes. The processing node removed the value of the primary key field or unique key from the data, resulting in an error.",
+			describeCN = "数据经过主从合并节点时，数据中不存在主键或唯一索引的值，可能的原因\n" +
+					"在主从合并节点前、源节点后，存在处理节点，如JS节点。处理节点将数据中的主键字段或唯一键的值移除，导致错误。",
+			dynamicDescription = "Data content: {}, primary or unique key Field: {}",
+			dynamicDescriptionCN = "数据内容：{}，主键或唯一键字段：{}",
+			solution = "If there are other processing nodes such as JS nodes before the master-slave merge node, check the script for code to remove the associated key from the data or remove the processing node after the source node",
+			solutionCN = "如果主从合并节点前存在其他处理节点如JS节点，请检查脚本中是否存在将数据中的关联键移除的代码或将源节点后继的处理节点删除"
 	)
 	String PK_OR_UNIQUE_VALUE_NOT_EXISTS = "16021";
 
@@ -140,32 +214,47 @@ public interface TaskMergeProcessorExCode_16 {
 	String NOT_FOUND_SOURCE_NODE = "16022";
 
 	@TapExCode(
-			describe = "The table name cannot be blank",
-			describeCN = "表名不能为空"
+			describe = "The node name of a node of type TableNode is detected as null",
+			describeCN = "检测出TableNode类型节点的节点名为空"
 	)
 	String TABLE_NAME_CANNOT_BE_BLANK = "16023";
 
 	@TapExCode(
 			describe = "The connection id cannot be blank",
-			describeCN = "连接id不能为空"
+			describeCN = "检测出TableNode类型节点连接id为空",
+			dynamicDescription = "Table Node info：{}",
+			dynamicDescriptionCN = "节点的信息：{}"
 	)
 	String CONNECTION_ID_CANNOT_BE_BLANK = "16024";
+	//TODO 下面的不能合并也应该封装错误码
 
 	@TapExCode(
-			describe = "The merged source table node not found, please check you config",
-			describeCN = "合并的源表节点不存在，请检查你的配置"
+			describe = "It failed to obtain the pre-node information of the master-slave merge based on the node ID",
+			describeCN = "根据节点ID获取主从合并的前置节点信息失败"
 	)
 	String TAP_MERGE_TABLE_NODE_NOT_FOUND = "16025";
 
 	@TapExCode(
-			describe = "Updating or writing to the cache, finding data from the cache based on the federated key failed",
-			describeCN = "批量更新或写入缓存，根据联合键从缓存中查找数据失败"
+			describe = "Batch update or write to the cache, lookup data from the cache based on the union key failed",
+			describeCN = "批量更新或写入缓存，根据联合键从缓存中查找数据失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"The external storage database type configured is :{}, the external storage name is :{}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，关联键值：{}，编码后的关联键值：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String UPSERT_CACHE_FIND_BY_JOIN_KEYS_FAILED = "16026";
 
 	@TapExCode(
 			describe = "Update or write cache failed",
-			describeCN = "批量更新或写入缓存失败"
+			describeCN = "批量更新或写入缓存失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Node name: {}, table name: {}, encoded association key value: {}, configured external storage database type: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，表名：{}，编码后的关联键值：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String UPSERT_CACHES_FAILED = "16027";
 	@TapExCode(
@@ -201,19 +290,25 @@ public interface TaskMergeProcessorExCode_16 {
 	String CHECK_UPDATE_JOIN_KEY_VALUE_CACHE_FAILED_CANT_GET_SOURCE_CONNECTION_BY_ID = "16033";
 
 	@TapExCode(
-			describe = "Failed to establish a cache to detect whether the join key value has changed. The node id is empty and the name cannot be constructed",
-			describeCN = "建立检测关联键值是否变更缓存失败，节点id为空，无法构建名称"
+			describe = "The establishment of the cache to detect whether the associated key value changes fails, and the node id is empty, and the name of the cache cannot be constructed according to the id",
+			describeCN = "建立检测关联键值是否变更缓存失败，检查出节点id为空，无法根据id构建缓存的名称"
 	)
 	String GET_CHECK_UPDATE_JOIN_KEY_VALUE_CACHE_NAME_FAILED_NODE_ID_CANNOT_NULL = "16034";
 
 	@TapExCode(
-			describe = "Failed to establish a cache to detect whether the join key value has changed, and the write signature failed",
-			describeCN = "建立检测关联键值是否变更缓存失败，写入签名失败"
+			describe = "Failed to establish a write signature that checks whether the associated key value changes the cache",
+			describeCN = "建立检测关联键值变更缓存时，写入签名失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Cache name: {}, signature content: {}, external storage database type configured: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"缓存名称：{}，签名内容：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String INIT_CHECK_UPDATE_JOIN_KEY_VALUE_CACHE_WRITE_SIGN_FAILED = "16035";
 	@TapExCode(
 			describe = "Failed to copy external storage, original dto is null",
-			describeCN = "复制外部存储失败，原始对象为空"
+			describeCN = "复制外部存储失败，检测出原始外部存储对象为空"
 	)
 	String COPY_EXTERNAL_STORAGE_FAILED_SOURCE_IS_NULL = "16036";
 	@TapExCode(
@@ -226,9 +321,12 @@ public interface TaskMergeProcessorExCode_16 {
 			describeCN = "分析引用关系失败，合并配置字典为空，可能是因为调用顺序错误"
 	)
 	String ANALYZE_REFERENCE_FAILED_TABLE_MERGE_MAP_EMPTY = "16038";
+	//TODO
 	@TapExCode(
 			describe = "Failed to get and update the join key cache, unable to get the IMAP",
-			describeCN = "获取并更新关联键缓存失败，无法获取IMAP"
+			describeCN = "根据节点id获取关联键值变更缓存失败",
+			dynamicDescription = "Node ID: {}, Node Name: {}",
+			dynamicDescriptionCN = "节点ID：{}，Node Name：{}"
 	)
 	String GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_GET_IMAP = "16039";
 	@TapExCode(
@@ -238,12 +336,24 @@ public interface TaskMergeProcessorExCode_16 {
 	String GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_AFTER_IS_EMPTY = "16040";
 	@TapExCode(
 			describe = "Failed to get and update the join key cache, failed to find by PK",
-			describeCN = "获取并更新关联键缓存失败，根据PK查找失败"
+			describeCN = "根据主键或唯一键查找关联键缓存失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Node name: {}, primary key or unique key value: {}, external storage database type configured: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，主键或唯一键值：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_FIND_BY_PK_FAILED = "16041";
 	@TapExCode(
 			describe = "Failed to get and update the join key cache, upsert failed",
-			describeCN = "获取并更新关联键缓存失败，更新after数据失败"
+			describeCN = "更新关联键缓存中数据失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"Node name: {}, primary key or unique key value: {}, after content: {}, external storage database type configured: {}, external storage name: {}, storage structure name: {}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"节点名称：{}，主键或唯一键值：{}，after的内容：{}，配置的外存数据库类型为：{}，外存名称为：{}，存储结构名称：{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. Find the external storage configuration configured by the master-slave merge node, and conduct a connection test to see if it is available. If not, you need to troubleshose according to the test failure information. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_UPSERT_FAILED = "16042";
 	@TapExCode(
@@ -347,7 +457,13 @@ public interface TaskMergeProcessorExCode_16 {
 
 	@TapExCode(
 			describe = "Init Cache failed",
-			describeCN = "初始化缓存失败"
+			describeCN = "由于主从合并功能需要使用缓存，启动带有主从合并节点的任务时，根据主从合并节点配置中的外存配置初始化缓存失败",
+			dynamicDescription = "The configured external cache database is unable to provide normal service\n" +
+					"The external storage database type configured is :{}, the external storage name is :{}",
+			dynamicDescriptionCN = "配置的外存缓存数据库无法正常提供服务\n" +
+					"配置的外存数据库类型为:{}，外存名称为:{}",
+			solution = "To check whether the external storage cache database is normal, you can click the sub-menu \"external storage Management\" in the parent menu \"System Management\" in the left menu bar of the system to enter the external storage management interface. The external memory configuration corresponding to the shared mining task is found, and the connection test is carried out to see if it is available. If it is not available, the test failure information is used to troubleshoate. Restart the task after the connection test passes",
+			solutionCN = "检查外存缓存数据库是否正常，可以点击系统左侧菜单栏中的父菜单“系统管理”中的子菜单“外存管理”进入外存管理界面。找到主从合并节点配置的外存配置，进行连接测试看是否可用，如不可用则需要根据测试失败信息进行排查。待连接测试通过后重新启动任务"
 	)
 	String INIT_CACHE_FAILED = "16062";
 }
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 294371345d..d47bef5a52 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
@@ -19,7 +19,7 @@ public interface TaskTargetProcessorExCode_15 {
 			dynamicDescription = "Failed to get table model with table name {}",
 			dynamicDescriptionCN = "获取表名为 {} 的表模型失败",
 			solution = "On the management side, reset the task and run the task again to see if the task can be started normally. If it cannot be started normally, you need to raise a work order to troubleshose the problem",
-			solutionCN = "1、在管理端重置任务并重新运行任务，看任务能否正常启动，如不能正常启动则需要提工单排查问题"
+			solutionCN = "在管理端重置任务并重新运行任务，看任务能否正常启动，如不能正常启动则需要提工单排查问题"
 	)
 	String INIT_TARGET_TABLE_TAP_TABLE_NULL = "15002";
 
@@ -55,8 +55,8 @@ public interface TaskTargetProcessorExCode_15 {
 					"删除表失败，表名为：{}",
 			solution = "1. the target node connection to use user delete table permissions\n"+
 						"2. Remove the foreign key constraint from the associated table and restart the task",
-			solutionCN = "1、授予目标节点连接使用用户删除表权限\n"+
-					"2、先删除关联表的外键约束后重新启动任务"
+			solutionCN = "1. 授予目标节点连接使用用户删除表权限\n"+
+					"2. 先删除关联表的外键约束后重新启动任务"
 	)
 	String DROP_TABLE_FAILED = "15004";
 
@@ -146,9 +146,7 @@ public interface TaskTargetProcessorExCode_15 {
 	)
 	String ALTER_FIELD_ATTR_CANNOT_GET_TAP_TABLE = "15013";
 	@TapExCode(
-			describe = "The target node fails to modify the field attribute, and the reason is analyzed: \n"+
-					"1. the event of modifying the field attribute is repeatedly executed by the target node. If the target has executed the field attribute modification operation, the task is started in the \"incremental\" synchronous way after stopping the task and the incremental acquisition starts before the field attribute modification event, the target node executes the field repeatedly and an error is reported"+
-					"2. The account used by the target node to connect does not have the corresponding Alter permission, resulting in the failure to modify the field name",
+			describe = "The target node failed to perform the operation to modify the field attribute",
 			describeCN = "目标节点执行修改字段属性操作时失败",
 			dynamicDescription = "1. The change field property event is repeatedly executed by the destination node. If the target has executed the field attribute modification operation, the task is started in the \"incremental\" synchronous way after stopping the task and the incremental acquisition starts before the field attribute modification event, the target node executes the field repeatedly and an error is reported\n" +
 					"2. The account that the destination node uses to connect does not have the corresponding permission, resulting in the failure to modify the field attribute\n" +
@@ -164,16 +162,16 @@ public interface TaskTargetProcessorExCode_15 {
 	String ALTER_FIELD_ATTR_EXECUTE_FAILED = "15014";
 	@TapExCode(
 			describe = "The target node fails to delete the field operation",
-			describeCN = "The target node failed to perform the delete field operation",
-			dynamicDescription = "1. The deletion event is repeatedly executed by the destination node. If the target has executed the field attribute modification operation, the task is started in the \"incremental\" synchronous way after stopping the task and the incremental acquisition starts before the field attribute modification event, the target node executes the field repeatedly and an error is reported\n" +
-					"2. The account used by the target node to connect does not have the corresponding permission, resulting in the failure to delete the field\n" +
+			describeCN = "目标节点执行删除字段操作时失败",
+			dynamicDescription = "1. The deletion event is repeatedly executed by the destination node. If the target has executed the deletion operation, the task is started in the \"incremental\" synchronous way after stopping the task and the incremental acquisition starts before the deletion field attribute event, the target node will repeatedly execute the deletion field operation and an error will be reported\n" +
+					"2. The account that the destination node uses to connect does not have the corresponding permission, resulting in the failure to delete the field\n" +
 					"Table name for deletion event :{}, field name for deletion field :{}",
-			dynamicDescriptionCN = "1. 删除事件被目标节点重复执行。如果目标已经执行了修改字段属性操作后，停止任务后以“增量”的同步方式启动任务并且增量采集开始时刻在修改字段属性事件之前，导致目标节点重复执行字段而发生报错\n" +
+			dynamicDescriptionCN = "1. 删除事件被目标节点重复执行。如果目标已经执行了删除性操作后，停止任务后以“增量”的同步方式启动任务并且增量采集开始时刻在删除字段属性事件之前，导致目标节点重复执行删除字段操作而发生报错\n" +
 					"2. 目标节点使用连接的账号没有相应的权限，导致删除字段失败\n" +
 					"删除事件的表名:{}，删除字段的字段名:{}",
 			solution = "1. When the task is started in the \"incremental\" synchronous way, the specified time point of deleting the field event needs to be repaired at the beginning of the specified incremental collection time, so as to avoid the error caused by the target node repeatedly executing deleting the field operation\n" +
 					"2. Grant the destination node the appropriate permissions to use the connected account",
-			solutionCN = "1. 以“增量”的同步方式启动任务时，指定的增量采集开始时刻需要再修删除字段事件的时间点后，避免目标节点重复执行删除字段操作发生报错\n"+
+			solutionCN = "1. 以“增量”的同步方式启动任务时，指定的增量采集开始时刻需要在删除字段事件的时间点后，避免目标节点重复执行删除字段操作发生报错\n"+
 					"2. 授予目标节点使用连接的账号相应的权限"
 	)
 	String DROP_FIELD_EXECUTE_FAILED = "15015";
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/exception/TapExactlyOnceWriteExCode_22.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/exception/TapExactlyOnceWriteExCode_22.java
index b1fbffc921..7715c248cb 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/exception/TapExactlyOnceWriteExCode_22.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/exception/TapExactlyOnceWriteExCode_22.java
@@ -15,7 +15,7 @@ public interface TapExactlyOnceWriteExCode_22 {
 	String UNKNOWN_ERROR = "22001";
 
 	@TapExCode(
-			describe = "When the target initiates incremental exact-once writes, the engine generates a exact-once cache table (which records incremental events that have been written to the target) and creates it in the target database. This error is raised when the exact cache table structure for the target database fails to be generated",
+			describe = "When the target initiates incremental exact-once writes, the engine generates a exact-once cache table (which records incremental events that have been written to the target) and creates it in the target database. Failed to generate an accurate cache table structure according to the target database",
 			describeCN = "当目标开启增量精确一次写入时，引擎会生成一张精准一次缓存表（用于记录已经写入目标的增量事件）并在目标数据库创建。根据目标数据库生成精准一次缓存表结构失败",
 			dynamicDescription = "Failed to generate the target database table based on the target database type with the fields: {}",
 			dynamicDescriptionCN = "根据目标数据库类型生成目标数据库表时失败，表的字段为: {}"
@@ -23,16 +23,16 @@ public interface TapExactlyOnceWriteExCode_22 {
 	String TARGET_TYPES_GENERATOR_FAILED = "22002";
 
 	@TapExCode(
-			describe = "When the target initiates incremental exact-once write, each incremental event generates a corresponding precision-once record event written to the exact-once cache table, and they are written to the database in the same transaction. Failed to write exactly one record event. And it rolls back the whole batch of events",
-			describeCN = "当目标开启增量精确一次写入时，每条增量事件会生成对应的一条精准一次写的记录事件写入精确一次缓存表，并且它们是在同一个事务写入数据库。精准一次记录事件写入目标数据库失败。并且会回滚整批事件",
+			describe = "When the target initiates incremental exact-once write, each incremental event generates a corresponding precision-once record event written to the exact-once cache table, and they are written to the database in the same transaction. Failed to write an accurate record event to the target database. And rolls back the whole batch of events",
+			describeCN = "当目标开启增量精确一次写入时，每条增量事件会生成对应的一条精准一次写的记录事件写入精确一次缓存表，并且它们是在同一个事务写入数据库。精准一次记录事件写入目标数据库时失败，并且会回滚整批事件",
 			dynamicDescription = "1. The target database is not available \n" +
 					"2. The cache table (named _TAP_EXACTLY_ONCE_CACHE) is deleted exactly once\n",
 			dynamicDescriptionCN = "1. 目标数据库无法提供正常服务\n" +
 					"2. 精确一次缓存表(表名为_TAP_EXACTLY_ONCE_CACHE)被删除\n",
 			solution = "1. Check whether it is the target database problem according to the cause of failure. If there is an outage, you need to restore the target database and restart the synchronization task\n" +
-					"2、Check the database's exact one-time cache table (the table name is _TAP_EXACTLY_ONCE_CACHE) to see if it is deleted by mistake. If it is deleted by mistake, restart the synchronization task",
-			solutionCN = "1、根据失败原因排查是否为目标数据库问题。如是否为目标数据库宕机，如出现宕机，则需要恢复目标数据库并重新启动同步任务\n" +
-					"2、查看数据库精确一次缓存表(表名为_TAP_EXACTLY_ONCE_CACHE)是否被误删，如被误删，则需要重新启动同步任务"
+					"2. Check the database's exact one-time cache table (the table name is _TAP_EXACTLY_ONCE_CACHE) to see if it is deleted by mistake. If it is deleted by mistake, restart the synchronization task",
+			solutionCN = "1. 根据失败原因排查是否为目标数据库问题。如是否为目标数据库宕机，如出现宕机，则需要恢复目标数据库并重新启动同步任务\n" +
+					"2. 查看数据库精确一次缓存表(表名为_TAP_EXACTLY_ONCE_CACHE)是否被误删，如被误删，则需要重新启动同步任务"
 	)
 	String WRITE_CACHE_FAILED = "22003";
 
@@ -57,8 +57,7 @@ public interface TapExactlyOnceWriteExCode_22 {
 	String WRITE_CACHE_FAILED_TIMESTAMP_IS_NULL = "22006";
 
 	@TapExCode(
-			describe = "When the target initiates the incremental exact write, it will query the exact cache table for whether the incremental event has been written according to the record event of the exact write before the incremental write. This error is raised when an exception occurs while retrieving the exact cached table. Explanation: \n" +
-					"1. Check whether the exact cache table is deleted in the target library",
+			describe = "When the target initiates the incremental exact write, it will query the exact cache table for whether the incremental event has been written according to the record event of the exact write before the incremental write. Failed to query the cache table exactly once",
 			describeCN = "当目标开启增量精确一次写入时，在增量写入前，会根据精确一次写的记录事件去精确一次缓存表中查询是否增量事件已经被写入。查询精确一次缓存表时失败",
 			dynamicDescription = "The query for the exact cache table failed, the query condition is: {}, the exact table name is: {}",
 			dynamicDescriptionCN = "查询精确一次缓存表失败，查询条件为：{}，查询精确一次的表名为：{}"
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
index 5290bf8bb7..d4834f9477 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
@@ -156,6 +156,7 @@
  **/
 public class HazelcastSourcePdkDataNode extends HazelcastSourcePdkBaseNode {
 	private static final String TAG = HazelcastSourcePdkDataNode.class.getSimpleName();
+	public static final int DEFAULT_POLLING_CDC_HEART_BEAT_TIME = 2;
 	//	private final Logger logger = LogManager.getLogger(HazelcastSourcePdkDataNode.class);
 	private final Logger logger = LogManager.getRootLogger();
 	private static final int CDC_POLLING_MIN_INTERVAL_MS = 500;
@@ -1116,6 +1117,7 @@ private void doPollingCDC() {
 		long logLoopTime = logInterval / cdcPollingInterval;
 		long heartbeatInterval = TimeUnit.MINUTES.toMillis(1);
 		long heartbeatTime = heartbeatInterval / cdcPollingInterval;
+		heartbeatTime = Math.max(heartbeatTime, DEFAULT_POLLING_CDC_HEART_BEAT_TIME);
 		int cdcPollingBatchSize = ((TableNode) node).getCdcPollingBatchSize();
 		cdcPollingBatchSize = Math.max(cdcPollingBatchSize, CDC_POLLING_MIN_BATCH_SIZE);
 
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastCustomProcessor.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastCustomProcessor.java
index a668c70648..6a195ee818 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastCustomProcessor.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastCustomProcessor.java
@@ -84,7 +84,8 @@ protected void doInit(@NotNull Context context) throws TapCodeException {
 				stateMap = getStateMap(context.hazelcastInstance(), node.getId());
 				((ScriptEngine) engine).put("state", stateMap);
 			} catch (ScriptException e) {
-				throw new TapCodeException(ScriptProcessorExCode_30.CUSTOM_PROCESSOR_GET_SCRIPT_ENGINE_FAILED, "Init script engine failed", e);
+				throw new TapCodeException(ScriptProcessorExCode_30.CUSTOM_PROCESSOR_GET_SCRIPT_ENGINE_FAILED, "Init script engine failed", e)
+						.dynamicDescriptionParameters(e.getMessage());
 			}
 		}
 	}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJavaScriptProcessorNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJavaScriptProcessorNode.java
index f5bbe43226..5c11311e2c 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJavaScriptProcessorNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastJavaScriptProcessorNode.java
@@ -160,7 +160,8 @@ protected Invocable getOrInitEngine() {
 						new ObsScriptLogger(getScriptObsLogger(), logger),
 						this.standard);
 			} catch (ScriptException e) {
-				throw new TapCodeException(ScriptProcessorExCode_30.JAVA_SCRIPT_PROCESSOR_GET_SCRIPT_FAILED, e);
+				throw new TapCodeException(ScriptProcessorExCode_30.JAVA_SCRIPT_PROCESSOR_GET_SCRIPT_FAILED, e)
+						.dynamicDescriptionParameters(e.getMessage());
 			}
 			if (!this.standard) {
 				this.scriptExecutorsManager = new ScriptExecutorsManager(new ObsScriptLogger(getScriptObsLogger()), clientMongoOperator, jetContext.hazelcastInstance(),
@@ -276,7 +277,7 @@ protected void tryProcess(TapdataEvent tapdataEvent, BiConsumer<TapdataEvent, Pr
 				thread.interrupt();
 			}
 			if (errorAtomicRef.get() != null) {
-				throw new TapCodeException(ScriptProcessorExCode_30.JAVA_SCRIPT_PROCESS_FAILED, errorAtomicRef.get());
+				throw new TapCodeException(ScriptProcessorExCode_30.JAVA_SCRIPT_PROCESS_FAILED, errorAtomicRef.get().getMessage());
 			}
 
 		} else {
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 a52969a002..d8c21a3fbb 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
@@ -541,7 +541,7 @@ protected void initCheckJoinKeyUpdateCacheMap() {
 		}
 	}
 
-	private void putInCheckJoinKeyUpdateCacheMapAndWriteSign(String id, String cacheName) {
+	protected void putInCheckJoinKeyUpdateCacheMapAndWriteSign(String id, String cacheName) {
 		List<String> joinKeyIncludePK = checkJoinKeyIncludePK(id);
 		if (CollectionUtils.isNotEmpty(joinKeyIncludePK) && !isSourceHaveBefore(id)) {
 			throw new TapCodeException(TaskMergeProcessorExCode_16.BUILD_CHECK_UPDATE_JOIN_KEY_CACHE_FAILED_JOIN_KEY_INCLUDE_PK, String.format("Join key include pk, id: %s, both join key and pk: %s", id, joinKeyIncludePK));
@@ -779,7 +779,8 @@ protected static ConstructIMap<Document> checkBuildConstructIMap(HazelcastInstan
 		try{
 			constructIMapV2 = buildConstructIMap(hazelcastInstance, referenceId, String.valueOf(cacheName.hashCode()), externalStorageDtoCopy);
 		}catch (Exception e) {
-			throw new TapCodeException(TaskMergeProcessorExCode_16.INIT_CACHE_FAILED, String.format("Cache name: %s", cacheName), e);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.INIT_CACHE_FAILED, String.format("Cache name: %s", cacheName), e)
+					.dynamicDescriptionParameters(externalStorageDtoCopy.getType(),externalStorageDtoCopy.getName());
 		}
 		if(constructIMapV2.isEmpty()){
 			ConstructIMap<Document> constructIMapV1;
@@ -915,13 +916,15 @@ protected void initSourcePkOrUniqueFieldMap(List<MergeTableProperties> mergeTabl
 					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));
+							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))
+									.dynamicDescriptionParameters(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));
+							throw new TapCodeException(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_ARRAY_KEY, String.format("- Table name: %s- Node name: %s\n", tableName, nodeName))
+									.dynamicDescriptionParameters(tableName,nodeName);
 						}
 						fieldNames = arrayKeys;
 						break;
@@ -1222,8 +1225,11 @@ protected void upsertCache(TapdataEvent tapdataEvent, MergeTableProperties merge
 		try {
 			groupByJoinKeyValues = hazelcastConstruct.find(encodeJoinValueKey);
 		} catch (Exception e) {
+			Node preNode = getPreNode(mergeTableProperty.getId());
 			throw new TapEventException(TaskMergeProcessorExCode_16.UPSERT_CACHE_FIND_BY_JOIN_KEY_FAILED, String.format("- Construct name: %s\n- Join value key: %s, encode: %s",
-					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e).addEvent(tapdataEvent.getTapEvent());
+					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e).addEvent(tapdataEvent.getTapEvent())
+					.dynamicDescriptionParameters(preNode.getName(), getTableName(tapdataEvent), joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey,
+							encodePkOrUniqueValueKey, tapdataEvent.getTapEvent(), externalStorageDto.getType(), externalStorageDto.getName(), hazelcastConstruct.getName());
 		}
 		if (null == groupByJoinKeyValues) {
 			groupByJoinKeyValues = new Document();
@@ -1233,8 +1239,12 @@ protected void upsertCache(TapdataEvent tapdataEvent, MergeTableProperties merge
 		try {
 			hazelcastConstruct.upsert(encodeJoinValueKey, groupByJoinKeyValues);
 		} catch (Exception e) {
+			String preTableName = getPreTableName(tapdataEvent);
+			Node preNode = getPreNode(mergeTableProperty.getId());
 			throw new TapEventException(TaskMergeProcessorExCode_16.UPSERT_CACHE_FAILED, String.format("- Construct name: %s\n- Join value key: %s, encode: %s\n- Pk or unique values: %s, encode: %s\n- Find by join value key result: %s",
-					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey, encodePkOrUniqueValueKey, groupByJoinKeyValues.toJson()), e).addEvent(tapdataEvent.getTapEvent());
+					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey, encodePkOrUniqueValueKey, groupByJoinKeyValues.toJson()), e).addEvent(tapdataEvent.getTapEvent())
+					.dynamicDescriptionParameters(preNode.getName(), preTableName, joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey, encodePkOrUniqueValueKey,
+							groupByJoinKeyValues.toJson(), externalStorageDto.getType(), externalStorageDto.getName(), hazelcastConstruct.getName());
 		}
 	}
 
@@ -1255,7 +1265,10 @@ protected void upsertCache(List<TapdataEvent> tapdataEvents, MergeTablePropertie
 				Thread.currentThread().interrupt();
 				return;
 			}
-			throw new TapCodeException(TaskMergeProcessorExCode_16.UPSERT_CACHE_FIND_BY_JOIN_KEYS_FAILED, "Find by keys: " + joinValueKeyTapdataEventMap.keySet() + ", first event: " + tapdataEvents.get(0).getTapEvent(), e);
+			Node preNode = getPreNode(mergeTableProperties.getId());
+			String tableName = getTableName(preNode);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.UPSERT_CACHE_FIND_BY_JOIN_KEYS_FAILED, "Find by keys: " + joinValueKeyTapdataEventMap.keySet() + ", first event: " + tapdataEvents.get(0).getTapEvent(), e)
+					.dynamicDescriptionParameters(preNode.getName(), tableName,joinValueKeyTapdataEventMap.keySet(), joinValueKeyTapdataEventMap, externalStorageDto.getType(), externalStorageDto.getName(), hazelcastConstruct.getName());
 		}
 		Map<String, Document> insertMap = new HashMap<>();
 		for (String joinValueKey : joinValueKeyTapdataEventMap.keySet()) {
@@ -1280,7 +1293,10 @@ protected void upsertCache(List<TapdataEvent> tapdataEvents, MergeTablePropertie
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
 		} catch (Exception e) {
-			throw new TapCodeException(TaskMergeProcessorExCode_16.UPSERT_CACHES_FAILED, e);
+			Node preNode = getPreNode(mergeTableProperties.getId());
+			String tableName = getTableName(preNode);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.UPSERT_CACHES_FAILED, e)
+					.dynamicDescriptionParameters(preNode.getName(), tableName, groupByJoinKeyValues.keySet(), externalStorageDto.getType(), externalStorageDto.getName(), hazelcastConstruct.getName());
 		}
 	}
 
@@ -1293,7 +1309,7 @@ protected void transformDateTime(Map<String, Object> after) {
 		});
 	}
 
-	private void deleteCache(TapdataEvent tapdataEvent, MergeTableProperties mergeTableProperty, ConstructIMap<Document> hazelcastConstruct) throws Exception {
+	protected void deleteCache(TapdataEvent tapdataEvent, MergeTableProperties mergeTableProperty, ConstructIMap<Document> hazelcastConstruct) throws Exception {
 		Map<String, Object> before = getBefore(tapdataEvent);
 		String joinValueKey = getJoinValueKeyBySource(before, mergeTableProperty, hazelcastConstruct);
 		String encodeJoinValueKey = encode(joinValueKey);
@@ -1303,8 +1319,12 @@ private void deleteCache(TapdataEvent tapdataEvent, MergeTableProperties mergeTa
 		try {
 			groupByJoinKeyValues = hazelcastConstruct.find(encodeJoinValueKey);
 		} catch (Exception e) {
+			Node preNode = getPreNode(mergeTableProperty.getId());
+			String tableName = getTableName(preNode);
 			throw new TapEventException(TaskMergeProcessorExCode_16.DELETE_CACHE_FIND_BY_JOIN_KEY_FAILED, String.format("- Construct name: %s\n- Join value key: %s, encode: %s",
-					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e).addEvent(tapdataEvent.getTapEvent());
+					hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e).addEvent(tapdataEvent.getTapEvent())
+					.dynamicDescriptionParameters(preNode.getName(),tableName,joinValueKey,encodeJoinValueKey,pkOrUniqueValueKey,
+							encodePkOrUniqueValueKey,externalStorageDto.getType(),externalStorageDto.getName(),hazelcastConstruct.getName());
 		}
 		if (null == groupByJoinKeyValues) {
 			return;
@@ -1314,8 +1334,12 @@ private void deleteCache(TapdataEvent tapdataEvent, MergeTableProperties mergeTa
 			try {
 				hazelcastConstruct.delete(encodeJoinValueKey);
 			} catch (Exception e) {
+				Node preNode = getPreNode(mergeTableProperty.getId());
+				String tableName = getTableName(preNode);
 				throw new TapEventException(TaskMergeProcessorExCode_16.DELETE_CACHE_FAILED, String.format("- Construct name: %s\n- Join value key: %s, encode: %s\n- Pk or unique value key: %s, encode: %s\n- Find by join value key result: %s",
-						hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey, encodePkOrUniqueValueKey, groupByJoinKeyValues.toJson()), e).addEvent(tapdataEvent.getTapEvent());
+						hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey, pkOrUniqueValueKey, encodePkOrUniqueValueKey, groupByJoinKeyValues.toJson()), e).addEvent(tapdataEvent.getTapEvent())
+						.dynamicDescriptionParameters(preNode.getName(),tableName,joinValueKey,encodeJoinValueKey,pkOrUniqueValueKey,
+								encodePkOrUniqueValueKey,externalStorageDto.getType(),externalStorageDto.getName(),hazelcastConstruct.getName());
 			}
 		} else {
 			hazelcastConstruct.upsert(encodeJoinValueKey, groupByJoinKeyValues);
@@ -1332,7 +1356,7 @@ private Map<String, Object> getAfter(TapdataEvent tapdataEvent) {
 		return after;
 	}
 
-	private Map<String, Object> getBefore(TapdataEvent tapdataEvent) {
+	protected Map<String, Object> getBefore(TapdataEvent tapdataEvent) {
 		Map<String, Object> before = null;
 		if (tapdataEvent.getTapEvent() instanceof TapRecordEvent) {
 			before = TapEventUtil.getBefore(tapdataEvent.getTapEvent());
@@ -1342,11 +1366,12 @@ private Map<String, Object> getBefore(TapdataEvent tapdataEvent) {
 		return before;
 	}
 
-	private String getJoinValueKeyBySource(Map<String, Object> data, MergeTableProperties mergeProperty, ConstructIMap<Document> hazelcastConstruct) {
+	protected String getJoinValueKeyBySource(Map<String, Object> data, MergeTableProperties mergeProperty, ConstructIMap<Document> hazelcastConstruct) {
 		List<Map<String, String>> joinKeys = mergeProperty.getJoinKeys();
 		List<String> joinKeyList = getJoinKeys(joinKeys, JoinConditionType.SOURCE);
 		if (CollectionUtils.isEmpty(joinKeyList)) {
-			throw new TapCodeException(TaskMergeProcessorExCode_16.MISSING_SOURCE_JOIN_KEY_CONFIG, String.format("Map name: %s, Merge property: %s", hazelcastConstruct.getName(), mergeProperty));
+			throw new TapCodeException(TaskMergeProcessorExCode_16.MISSING_SOURCE_JOIN_KEY_CONFIG, String.format("Map name: %s, Merge property: %s", hazelcastConstruct.getName(), mergeProperty))
+					.dynamicDescriptionParameters(getPreNode(mergeProperty.getId()).getName(),mergeProperty);
 		}
 		if (MapUtils.isEmpty(data)) {
 			return "";
@@ -1355,7 +1380,9 @@ private String getJoinValueKeyBySource(Map<String, Object> data, MergeTablePrope
 		for (String joinKey : joinKeyList) {
 			Object value = MapUtilV2.getValueByKey(data, joinKey);
 			if (value instanceof NotExistsNode) {
-				throw new TapCodeException(TaskMergeProcessorExCode_16.JOIN_KEY_VALUE_NOT_EXISTS, String.format("- Map name: %s\n- Join key: %s\n- Data: %s", hazelcastConstruct.getName(), joinKey, data));
+				Node preNode = getPreNode(mergeProperty.getId());
+				throw new TapCodeException(TaskMergeProcessorExCode_16.JOIN_KEY_VALUE_NOT_EXISTS, String.format("- Map name: %s\n- Join key: %s\n- Data: %s", hazelcastConstruct.getName(), joinKey, data))
+						.dynamicDescriptionParameters(preNode.getName(),joinKey,data);
 			}
 			try {
 				values.add(convertJoinKeyValue2String(value));
@@ -1367,7 +1394,7 @@ private String getJoinValueKeyBySource(Map<String, Object> data, MergeTablePrope
 		return String.join("_", values);
 	}
 
-	private String getJoinValueKeyByTarget(Map<String, Object> data, MergeTableProperties mergeProperty, MergeTableProperties lastMergeProperty, ConstructIMap<Document> hazelcastConstruct) {
+	protected String getJoinValueKeyByTarget(Map<String, Object> data, MergeTableProperties mergeProperty, MergeTableProperties lastMergeProperty, ConstructIMap<Document> hazelcastConstruct) {
 		List<Map<String, String>> joinKeys = mergeProperty.getJoinKeys();
 		List<String> joinKeyList = getJoinKeys(joinKeys, JoinConditionType.TARGET);
 		if (CollectionUtils.isEmpty(joinKeyList)) {
@@ -1419,11 +1446,11 @@ public JoinKeyValueConvertNumberException(Throwable cause) {
 		}
 	}
 
-	private String encode(String str) {
+	protected String encode(String str) {
 		return Base64.getEncoder().encodeToString(str.getBytes(StandardCharsets.UTF_8));
 	}
 
-	private List<String> getJoinKeys(List<Map<String, String>> joinKeys, JoinConditionType joinConditionType) {
+	protected List<String> getJoinKeys(List<Map<String, String>> joinKeys, JoinConditionType joinConditionType) {
 		if (null == joinKeys) {
 			return Collections.emptyList();
 		}
@@ -1444,7 +1471,8 @@ protected String getPkOrUniqueValueKey(Map<String, Object> data, String sourceNo
 		for (String pkOrUniqueField : pkOrUniqueFields) {
 			Object value = MapUtilV2.getValueByKey(data, pkOrUniqueField);
 			if (value instanceof NotExistsNode) {
-				throw new TapCodeException(TaskMergeProcessorExCode_16.PK_OR_UNIQUE_VALUE_NOT_EXISTS, String.format("- Map name: %s\n- Pk or unique field: %s\n- Data: %s", hazelcastConstruct.getName(), pkOrUniqueFields, data));
+				throw new TapCodeException(TaskMergeProcessorExCode_16.PK_OR_UNIQUE_VALUE_NOT_EXISTS, String.format("- Map name: %s\n- Pk or unique field: %s\n- Data: %s", hazelcastConstruct.getName(), pkOrUniqueFields, data))
+						.dynamicDescriptionParameters(pkOrUniqueFields,data);
 			}
 			if (value instanceof Number) {
 				values.add(new BigDecimal(String.valueOf(value)).stripTrailingZeros().toPlainString());
@@ -1457,7 +1485,7 @@ protected String getPkOrUniqueValueKey(Map<String, Object> data, String sourceNo
 		return String.join("_", values);
 	}
 
-	private Node<?> getSourceTableNode(String sourceId) {
+	protected Node<?> getSourceTableNode(String sourceId) {
 		Node<?> node = this.processorBaseContext.getNode();
 		List<? extends Node<?>> predecessors = node.predecessors();
 		predecessors = predecessors.stream().filter(n -> n.getId().equals(sourceId)).collect(Collectors.toList());
@@ -1468,12 +1496,13 @@ private Node<?> getSourceTableNode(String sourceId) {
 		return predecessors.get(0);
 	}
 
-	private String getConnectionId(Node<?> preTableNode) {
+	protected String getConnectionId(Node<?> preTableNode) {
 		String connectionId;
 		if (preTableNode instanceof TableNode) {
 			connectionId = ((TableNode) preTableNode).getConnectionId();
 			if (StringUtils.isBlank(connectionId)) {
-				throw new TapCodeException(TaskMergeProcessorExCode_16.CONNECTION_ID_CANNOT_BE_BLANK, String.format("Table node: %s", preTableNode));
+				throw new TapCodeException(TaskMergeProcessorExCode_16.CONNECTION_ID_CANNOT_BE_BLANK, String.format("Table node: %s", preTableNode))
+						.dynamicDescriptionParameters();
 			}
 		} else {
 			throw new RuntimeException(preTableNode.getName() + "(" + preTableNode.getId() + ", " + preTableNode.getClass().getSimpleName() + ") cannot linked to a merge table node");
@@ -1540,7 +1569,8 @@ protected List<MergeLookupResult> recursiveLookup(MergeTableProperties mergeTabl
 						nodeLogger.debug("Lookup find data filter: {}({}), result: {}", joinValueKey, encodeJoinValueKey, findData);
 					}
 				} catch (Exception e) {
-					throw new TapCodeException(TaskMergeProcessorExCode_16.LOOK_UP_FIND_BY_JOIN_KEY_FAILED, String.format("- Find construct name: %s%n- Join key: %s%n- Encoded join key: %s", hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e);
+					throw new TapCodeException(TaskMergeProcessorExCode_16.LOOK_UP_FIND_BY_JOIN_KEY_FAILED, String.format("- Find construct name: %s%n- Join key: %s%n- Encoded join key: %s", hazelcastConstruct.getName(), joinValueKey, encodeJoinValueKey), e)
+							.dynamicDescriptionParameters(getPreNode(mergeTableProperties.getId()).getName(),getPreNode(childMergeProperty.getId()).getName(), joinValueKey, encodeJoinValueKey, externalStorageDto.getType(), externalStorageDto.getName(), hazelcastConstruct.getName());
 				}
 			}
 			io.tapdata.pdk.apis.entity.merge.MergeTableProperties copyMergeTableProperty = copyMergeTableProperty(childMergeProperty);
@@ -2076,7 +2106,8 @@ protected Map<String, Object> getAndUpdateJoinKeyCache(TapdataEvent tapdataEvent
 		ConstructIMap<Document> constructIMap = this.checkJoinKeyUpdateCacheMap.get(preNodeId);
 		if (null == constructIMap) {
 			Node<?> preNode = getPreNode(preNodeId);
-			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_GET_IMAP, "Node name: " + preNode.getName() + ", id: " + preNodeId);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_GET_IMAP, "Node name: " + preNode.getName() + ", id: " + preNodeId)
+					.dynamicDescriptionParameters(preNode.getName(),preNode.getId());
 		}
 		TapEvent tapEvent = tapdataEvent.getTapEvent();
 		Map<String, Object> after = TapEventUtil.getAfter(tapEvent);
@@ -2089,7 +2120,8 @@ protected Map<String, Object> getAndUpdateJoinKeyCache(TapdataEvent tapdataEvent
 		try {
 			beforeDoc = constructIMap.find(pkOrUniqueValueKey);
 		} catch (Exception e) {
-			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_FIND_BY_PK_FAILED, "Construct name: " + constructIMap.getName() + ", pk or unique value key: " + pkOrUniqueValueKey, e);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_FIND_BY_PK_FAILED, "Construct name: " + constructIMap.getName() + ", pk or unique value key: " + pkOrUniqueValueKey, e)
+					.dynamicDescriptionParameters(getPreNode(preNodeId).getName(),pkOrUniqueValueKey,  externalStorageDto.getType(), externalStorageDto.getName(),constructIMap.getName());
 		}
 		if (null == beforeDoc) {
 			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_FIND_BEFORE, String.format("Construct name: %s, after: %s, filter value: %s", constructIMap.getName(), after, pkOrUniqueValueKey));
@@ -2099,7 +2131,8 @@ protected Map<String, Object> getAndUpdateJoinKeyCache(TapdataEvent tapdataEvent
 		try {
 			constructIMap.upsert(pkOrUniqueValueKey, afterDoc);
 		} catch (Exception e) {
-			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_UPSERT_FAILED, "Construct name: " + constructIMap.getName() + ", pk or unique value key: " + pkOrUniqueValueKey + ", after: " + afterDoc.toJson(), e);
+			throw new TapCodeException(TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_UPSERT_FAILED, "Construct name: " + constructIMap.getName() + ", pk or unique value key: " + pkOrUniqueValueKey + ", after: " + afterDoc.toJson(), e)
+					.dynamicDescriptionParameters(pkOrUniqueValueKey, getPreNode(preNodeId).getName(), afterDoc.toJson(), externalStorageDto.getType(), externalStorageDto.getName(), constructIMap.getName());
 		}
 		return before;
 	}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorBaseNode.java
index c51caa284e..fd26a8064b 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorBaseNode.java
@@ -4,7 +4,6 @@
 import com.tapdata.entity.SyncStage;
 import com.tapdata.entity.TapdataEvent;
 import com.tapdata.entity.task.context.ProcessorBaseContext;
-import com.tapdata.exception.CloneException;
 import com.tapdata.tm.commons.dag.Node;
 import com.tapdata.tm.commons.dag.nodes.TableNode;
 import com.tapdata.tm.commons.dag.process.MigrateProcessorNode;
@@ -43,6 +42,8 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import static io.tapdata.observable.logging.ObsLogger.LOG_TAG_SOURCE_FOR_USER_SCRIPT;
+
 /**
  * @author samuel
  * @Description
@@ -591,7 +592,7 @@ protected String getTableName(Node<?> node) {
 	public ObsLogger getScriptObsLogger() {
 		if (this.scriptObsLogger == null) {
 			Set<String> tags = new HashSet<>();
-			tags.add("src=user_script");
+			tags.add(LOG_TAG_SOURCE_FOR_USER_SCRIPT);
 			List<String> otherTags = getLogTags();
 			if (CollectionUtils.isNotEmpty(otherTags)) {
 				tags.addAll(otherTags);
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/constant/CommonUtil.java b/iengine/iengine-common/src/main/java/com/tapdata/constant/CommonUtil.java
index 53a40b170c..b78bed4cd0 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/constant/CommonUtil.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/constant/CommonUtil.java
@@ -1,5 +1,7 @@
 package com.tapdata.constant;
 
+import com.tapdata.entity.values.BooleanNotExist;
+
 import java.math.BigDecimal;
 
 public class CommonUtil {
@@ -54,7 +56,18 @@ public static int compareObjects(Object[] val1, Object[] val2) {
 		for (int i = 0; i < len1; i++) {
 			Object obj1 = val1[i];
 			Object obj2 = val2[i];
-
+			if (obj1 instanceof Boolean && !(obj2 instanceof Boolean)) {
+				obj2 = toBoolean(obj2);
+				if(obj2 instanceof BooleanNotExist){
+					obj2 = null;
+				}
+			}
+			if (obj2 instanceof Boolean && !(obj1 instanceof Boolean)) {
+				obj1 = toBoolean(obj1);
+				if(obj1 instanceof BooleanNotExist){
+					obj1 = null;
+				}
+			}
 			if (obj1 == null && obj2 == null) {
 				continue; // Both are null, compare the next element
 			} else if (obj1 == null) {
@@ -88,4 +101,61 @@ public static int compareObjects(Object[] val1, Object[] val2) {
 
 		return 0; // Both arrays are equal
 	}
+
+	/**
+	 * @param val1
+	 * @param val2
+	 * @return 相等返回true，不相等返回false
+	 */
+	public static boolean compareBoolean(Boolean val1, Object val2) {
+		if (null == val1 && null == val2){
+			return true;
+		}
+		if (null == val1 || null == val2) {
+			return false;
+		}
+		if (val2 instanceof Boolean) {
+			return val1.equals(val2);
+		}
+		Object val2Boolean = CommonUtil.toBoolean(val2);
+		if (val2Boolean instanceof BooleanNotExist) {
+			return false;
+		}
+		return val1.equals(val2Boolean);
+	}
+
+
+	public static Object toBoolean(Object val) {
+		if (val instanceof Number) {
+			long intValue = ((Number) val).longValue();
+			if (intValue == 0) {
+				return false;
+			} else if (intValue == 1) {
+				return true;
+			} else {
+				return new BooleanNotExist();
+			}
+		}
+		if (val instanceof BigDecimal) {
+			BigDecimal bigDecimalValue = (BigDecimal) val;
+			if (bigDecimalValue.compareTo(BigDecimal.ZERO) == 0) {
+				return false;
+			} else if (bigDecimalValue.compareTo(BigDecimal.ONE) == 0) {
+				return true;
+			} else {
+				return new BooleanNotExist();
+			}
+		}
+		if (val instanceof String) {
+			String str = val.toString().toLowerCase().trim();
+			if ("0".equals(str) || "false".equals(str)) {
+				return false;
+			} else if ("1".equals(str) || "true".equals(str)) {
+				return true;
+			} else {
+				return new BooleanNotExist();
+			}
+		}
+		return new BooleanNotExist();
+	}
 }
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/JavaScriptFunctions.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/JavaScriptFunctions.java
index a8e5df4258..54b23241b9 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/entity/JavaScriptFunctions.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/JavaScriptFunctions.java
@@ -34,7 +34,14 @@ public JavaScriptFunctions() {
 	public String getJSFunction() {
 		if (isJar()) {
 			if (StringUtils.isNotBlank(function_name) && StringUtils.isNotBlank(className)) {
-				return String.format(JAVA_TYPE_TEMPLATE, function_name, className);
+				String name;
+				if (StringUtils.isNotBlank(className)) {
+					int index = className.split("\\.").length - 1;
+					name = className.split("\\.")[index];
+				}else{
+					name = function_name;
+				}
+				return String.format(JAVA_TYPE_TEMPLATE, name, className);
 			}
 		} else {
 			if (StringUtils.isNotBlank(function_name) && StringUtils.isNotBlank(function_body)) {
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/values/BooleanNotExist.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/values/BooleanNotExist.java
new file mode 100644
index 0000000000..54e25abb9f
--- /dev/null
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/values/BooleanNotExist.java
@@ -0,0 +1,4 @@
+package com.tapdata.entity.values;
+
+public class BooleanNotExist {
+}
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/ScriptUtil.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/ScriptUtil.java
index cff4e4bfb9..69228000d2 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/ScriptUtil.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/ScriptUtil.java
@@ -245,7 +245,7 @@ public static Object invokeScript(
 		Map<String, Object> eventMap = MapUtil.obj2Map(processContext.getEvent());
 		context.put("event", eventMap);
 		if (engine == null) {
-			throw new TapCodeException(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED,"script engine is null");
+			throw new TapCodeException(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED_ENGINE_NULL,"script engine is null");
 		}
 
 		((ScriptEngine) engine).put("context", context);
@@ -269,7 +269,8 @@ record = MapUtils.isNotEmpty(message.getBefore()) ? message.getBefore() : messag
 				o = engine.invokeFunction(functionName, record);
 			}
 		} catch (Exception e) {
-			throw new TapCodeException(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED,String.format("Invoke function %s error", functionName),e);
+			throw new TapCodeException(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED,String.format("Invoke function %s error", functionName),e)
+					.dynamicDescriptionParameters(e.getMessage());
 		}
 
 		return o;
@@ -380,7 +381,7 @@ public static String initBuildInMethod(List<JavaScriptFunctions> javaScriptFunct
 							if (clientMongoOperator instanceof HttpClientMongoOperator) {
 								File file = ((HttpClientMongoOperator) clientMongoOperator).downloadFile(null, "file/" + fileId, filePath.toString(), true);
 								if (null == file) {
-									throw new TapCodeException(ScriptProcessorExCode_30.INIT_BUILD_IN_METHOD_FAILED,String.format("file not found，fileId:%s,filePath:%s",fileId,filePath));
+									throw new TapCodeException(ScriptProcessorExCode_30.INIT_STANDARDIZATION_METHOD_FAILED,String.format("file not found，fileId:%s,filePath:%s",fileId,filePath));
 								}
 							} else {
 								GridFSBucket gridFSBucket = clientMongoOperator.getGridFSBucket();
@@ -391,7 +392,7 @@ public static String initBuildInMethod(List<JavaScriptFunctions> javaScriptFunct
 									Files.createFile(filePath);
 									Files.copy(gridFSDownloadStream, filePath, StandardCopyOption.REPLACE_EXISTING);
 								} catch (Exception e) {
-									throw new TapCodeException(ScriptProcessorExCode_30.INIT_BUILD_IN_METHOD_FAILED,String.format("create function jar file %s", filePath),e);
+									throw new TapCodeException(ScriptProcessorExCode_30.INIT_STANDARDIZATION_METHOD_FAILED,String.format("create function jar file %s", filePath),e);
 								}
 							}
 						}
@@ -399,7 +400,7 @@ public static String initBuildInMethod(List<JavaScriptFunctions> javaScriptFunct
 							URL url = filePath.toUri().toURL();
 							urlList.add(url);
 						} catch (Exception e) {
-							throw new TapCodeException(ScriptProcessorExCode_30.INIT_BUILD_IN_METHOD_FAILED,String.format("create function jar file %s", filePath), e);
+							throw new TapCodeException(ScriptProcessorExCode_30.INIT_STANDARDIZATION_METHOD_FAILED,String.format("create function jar file %s", filePath), e);
 						}
                     }
 				}
@@ -414,7 +415,8 @@ public static String initBuildInMethod(List<JavaScriptFunctions> javaScriptFunct
 	}
 
 	public static void urlClassLoader(Consumer<URLClassLoader> consumer, List<URL> urlList){
-		try(final URLClassLoader urlClassLoader = new CustomerClassLoader(urlList.toArray(new URL[0]), ScriptUtil.class.getClassLoader());) {
+		try {
+			final URLClassLoader urlClassLoader = new CustomerClassLoader(urlList.toArray(new URL[0]), ScriptUtil.class.getClassLoader());
 			if (consumer != null) {
 				consumer.accept(urlClassLoader);
 			}
@@ -505,7 +507,8 @@ public static Invocable getPyEngine(
 			e.put("tapLog", logger);
 			e.eval(globalScript);
 		} catch (Exception es){
-			throw new TapCodeException(ScriptProcessorExCode_30.GET_PYTHON_ENGINE_FAILED,String.format("Fail init python node,script eval %s error",e), es);
+			throw new TapCodeException(ScriptProcessorExCode_30.GET_PYTHON_ENGINE_FAILED,String.format("Fail init python node,script eval %s error",e), es)
+					.dynamicDescriptionParameters(es.getMessage());
 		}
 		evalImportSources(e, "");
 
@@ -547,7 +550,8 @@ public static String initPythonBuildInMethod(
 				consumer.accept(urlClassLoader);
 			}
 		}catch (IOException e){
-			throw new TapCodeException(ScriptProcessorExCode_30.INIT_PYTHON_METHOD_ERROR,String.format("Init python build in method failed,url:%s", urls[0]),e);
+			throw new TapCodeException(ScriptProcessorExCode_30.URL_CLASS_LOADER_ERROR,String.format("Init python build in method failed,url:%s", urls[0]),e)
+					.dynamicDescriptionParameters(urls[0]);
 		}
 		return  "import com.tapdata.constant.DateUtil as DateUtil\n" +
 				"import com.tapdata.constant.UUIDGenerator as UUIDGenerator\n" +
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/error/ScriptProcessorExCode_30.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/error/ScriptProcessorExCode_30.java
index 44af7b1567..ba579c4273 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/error/ScriptProcessorExCode_30.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/error/ScriptProcessorExCode_30.java
@@ -8,20 +8,33 @@ public interface ScriptProcessorExCode_30 {
     @TapExCode
     String GET_SCRIPT_ENGINE_ERROR = "30001";
     @TapExCode(
-            describe = "Loading class by URL failed",
-            describeCN = "通过URL加载类失败"
+            describe = "When initializing the Python script engine, the URL classloader failed to load the resources required by the Python engine",
+            describeCN = "初始化Python脚本引擎时，URL类加载器加载Python引擎需要的资源失败",
+            dynamicDescription = "Load resource path :{}",
+            dynamicDescriptionCN = "加载资源路径:{}",
+            solution = "1. Obtain the resource file under the resource path by contacting the after-sales service to obtain the data synchronization tool installation package or re-unzipping the existing installation package. Then put the resource file back to the corresponding directory path and restart",
+            solutionCN = "1. 通过联系售后获取数据同步工具安装包或重新解压已有的安装包获取资源路径下的资源文件。然后将资源文件放回到对应的目录路径下后重新启动"
     )
     String URL_CLASS_LOADER_ERROR = "30002";
     @TapExCode
     String INIT_PYTHON_METHOD_ERROR = "30003";
     @TapExCode(
-            describe = "Failed to call engine script",
-            describeCN = "调用引擎脚本失败"
+            describe = "The execution engine failed to execute the JavaScript script",
+            describeCN = "脚本引擎在执行脚本失败，可能造成失败的原因：\n" +
+                    "1. 脚本中存在语法错误或调用了未定义的方法\n" +
+                    "2. 脚本在执行的过程中出现异常",
+            dynamicDescription = "Error message if script execution failed :{}",
+            dynamicDescriptionCN = "执行脚本失败的错误信息:{}",
+            solution = "The task is restarted after checking the script according to the error message of the failed script execution and correcting it",
+            solutionCN = "根据执行脚本失败的错误信息检查脚本并修正后重新启动任务"
     )
     String INVOKE_SCRIPT_FAILED = "30004";
+
     @TapExCode(
-            describe = "Failed to initialize standard engine script build method",
-            describeCN = "初始化标准引擎脚本构建方法失败"
+            describe = "The standard JS node failed to load the custom function",
+            describeCN = "标准JS节点初始化时，加载函数管理中导入的第三方jar包失败",
+            solution = "Recompile the third-party jar and upload it again",
+            solutionCN = "将第三方jar包重新编译后重新上传"
     )
     String INIT_STANDARDIZATION_METHOD_FAILED = "30005";
     @TapExCode(
@@ -36,7 +49,7 @@ public interface ScriptProcessorExCode_30 {
     String INIT_BUILD_IN_METHOD_FAILED = "30007";
     @TapExCode(
             describe = "Engine eval source failed",
-            describeCN = "引擎评估源失败"
+            describeCN = "在初始化JS脚本引擎时，定义JS脚本引擎的函数失败"
     )
     String EVAL_SOURCE_ERROR = "30008";
     @TapExCode(
@@ -45,28 +58,26 @@ public interface ScriptProcessorExCode_30 {
     )
     String INIT_SCRIPT_ENGINE_FAILED = "30009";
     @TapExCode(
-            describe = "Apply classloader context failed",
-            describeCN = "应用类加载器上下文失败"
+            describe = "The function to define the script engine using the URL classloader failed",
+            describeCN = "使用URL类加载器定义脚本引擎的函数失败"
     )
     String APPLY_CLASS_LOADER_CONTEXT_FAILED = "30010";
-
     @TapExCode(
-            describe = "A syntax error in the Python script in the Python node of the task prevented the creation of the Python execution engine",
-            describeCN = "任务中的Python节点中的Python脚本中存在语法错误，导致无法创建Python执行引擎",
-            dynamicDescription = "A syntax error occurred in the Python script with the following error message :{}",
-            dynamicDescriptionCN = "Python脚本中存在语法错误，错误信息为:{}",
+            describe = "A syntax error in the script in the Python node prevented the Python script engine from being created",
+            describeCN = "Python节点中的脚本中存在语法错误，导致无法创建Python脚本引擎",
+            dynamicDescription = "Syntax error in Python script, error message :{}",
+            dynamicDescriptionCN = "Python脚本中存在语法错误，错误信息:{}",
             solution = "According to the error message, check and modify the script error code in the Python node, and restart the task after the trial run passes",
             solutionCN = "根据报错提示，检查并修改Python节点中的脚本错误代码，试运行通过后重新后重新启动任务"
     )
     String GET_PYTHON_ENGINE_FAILED = "30011";
-
     @TapExCode(
-            describe = "This error occurs when a standard JS or enhanced JS node in a task has an error while executing a JS script",
-            describeCN = "任务中的标准JS或增强JS节点在执行JS脚本时发生错误时，则会出现此报错",
+            describe = "An error occurred while the JS node was executing the script",
+            describeCN = "JS节点在执行脚本时发生错误",
             dynamicDescription = "There was an error in the JS script with the following message:{}",
             dynamicDescriptionCN = "JS脚本中存在错误，错误信息为:{}",
-            solution = "According to the error information, check and modify the code in the standard JS or enhanced JS script that causes the error, and restart the task after the trial run passes",
-            solutionCN = "根据报错信息，检查和修改标准JS或增强JS中脚本中导致错误代码，试运行通过后重新启动任务"
+            solution = "According to the error information, check and modify the code that causes the error in the JS node, and restart the task after the trial run passes",
+            solutionCN = "根据报错信息，检查和修改JS节点中导致错误代码，试运行通过后重新启动任务"
     )
     String JAVA_SCRIPT_PROCESS_FAILED = "30012";
     @TapExCode(
@@ -78,23 +89,29 @@ public interface ScriptProcessorExCode_30 {
             solutionCN = "根据报错信息，检查和Python脚本中的导致错误的代码，并试运行通过后重新启动任务"
     )
     String PYTHON_PROCESS_FAILED="30013";
-    String SCRIPT_IO="30014";
     @TapExCode(
-            describe = "A syntax error in the JS script in the JS node of the task prevented the creation of the JS script engine",
-            describeCN = "任务中的标准JS/增强JS节点中的JS脚本中存在语法错误，导致无法创建JS执行引擎",
-            dynamicDescription = "A syntax error occurred in the Python script with the following error message :{}",
+            describe = "There is a syntax error in the script in the JS node, and the JS script engine cannot be created",
+            describeCN = "JS节点中脚本中存在语法错误，导致无法创建JS脚本引擎",
+            dynamicDescription = "There is a syntax error in the JS script with the following message :{}",
             dynamicDescriptionCN = "JS脚本中存在语法错误，错误信息为:{}",
-            solution = "According to the error message, check and modify the script error code in the JS node, and restart the task after the trial run passes",
+            solution = "According to the error prompt, check and modify the script error code in the JS node, and restart the task after the trial run passes",
             solutionCN = "根据报错提示，检查并修改JS节点中的脚本错误代码，试运行通过后重新后重新启动任务"
     )
-    String JAVA_SCRIPT_PROCESSOR_GET_SCRIPT_FAILED="30015";
+    String JAVA_SCRIPT_PROCESSOR_GET_SCRIPT_FAILED="30014";
     @TapExCode(
             describe = "A syntax error in the js script in the custom js node of the task prevented the creation of the script engine",
-            describeCN = "任务中的自定义节点中的JS脚本中存在语法错误，导致无法创建JS执行引擎",
-            dynamicDescription = "A syntax error occurred in the js script with the following error message :{}",
-            dynamicDescriptionCN = "自定义JS脚本中存在语法错误，错误信息为:{}",
+            describeCN = "自定义节点中的脚本中存在语法错误，导致无法创建JS脚本引擎",
+            dynamicDescription = "There is a syntax error in the custom script. The error message is:",
+            dynamicDescriptionCN = "自定义脚本中存在语法错误，错误信息为:{}",
             solution = "According to the error message, check and modify the script error code in the custom js node, and restart the task after the trial run passes",
             solutionCN = "根据报错提示，检查并修改自定义JS节点中的脚本错误代码，试运行通过后重新后重新启动任务"
     )
-    String CUSTOM_PROCESSOR_GET_SCRIPT_ENGINE_FAILED="30016";
+    String CUSTOM_PROCESSOR_GET_SCRIPT_ENGINE_FAILED="30015";
+
+    @TapExCode(
+            describe = "Failed to execute the script, detected that the script execution engine is empty",
+            describeCN = "执行脚本时失败，检测出脚本引擎为空"
+    )
+    String INVOKE_SCRIPT_FAILED_ENGINE_NULL ="30016";
+
 }
diff --git a/iengine/iengine-common/src/main/java/io/tapdata/error/TaskProcessorExCode_11.java b/iengine/iengine-common/src/main/java/io/tapdata/error/TaskProcessorExCode_11.java
index 09363e38b6..af80baf401 100644
--- a/iengine/iengine-common/src/main/java/io/tapdata/error/TaskProcessorExCode_11.java
+++ b/iengine/iengine-common/src/main/java/io/tapdata/error/TaskProcessorExCode_11.java
@@ -35,8 +35,8 @@ public interface TaskProcessorExCode_11 {
 	String PYTHON_PROCESS_FAILED = "11005";
 
 	@TapExCode(
-			describe = "When the task starts, the engine will initialize the custom node according to the created custom node template, when the initialization does not get the defined template information",
-			describeCN = "任务启动时，引擎会根据创建的自定义节点模版初始化自定义节点，当初始化时获取不到定义的模版信息时",
+			describe = "When the task starts, the engine initializes the custom node based on the custom node template created. Failed to get custom template information when initializing a custom node",
+			describeCN = "任务启动时，引擎会根据创建的自定义节点模版初始化自定义节点。当初始化时自定义节点时获取自定义的模版信息失败",
 			dynamicDescription = "The template for the custom node is removed before the task is started",
 			dynamicDescriptionCN = "自定义节点的模版在任务启动前被删除",
 			solution = "Click the Custom node of the Advanced functions sub-menu on the left side of the system to view the currently created custom template. If the custom template is deleted, you'll need to recreate the custom template and reconfigure the task",
@@ -57,7 +57,7 @@ public interface TaskProcessorExCode_11 {
 
 	@TapExCode(
 			describe = "Cannot get metadata cause qualified name is empty",
-			describeCN = "目标节点或处理节点识别到有DDL事件时，会根据DDL事件更新内存中维护的表模型。当根据DDL事件中的表名获取不到模型的唯一编码时，则会报此错误",
+			describeCN = "目标节点或处理节点识别到有DDL事件时，会根据DDL事件更新内存中维护的表模型。根据DDL事件中的表名获取不到模型的唯一编码",
 			dynamicDescription = "When getting a unique model code for the table name {}, the model code is empty",
 			dynamicDescriptionCN = "获取表名为 {} 的唯一模型编码时，模型编码为空"
 	)
@@ -120,8 +120,8 @@ public interface TaskProcessorExCode_11 {
 	@TapExCode(
 			describe = "When a DDL event is encountered, the task stops with an error",
 			describeCN = "遇到DDL事件时，任务停止报错",
-			dynamicDescription = "Since the DDL synchronization in the advanced configuration of the source node of the task is configured to \"task stops with error when encountering DDL\", this error is reported when a DDL event is identified",
-			dynamicDescriptionCN = "由于任务的源节点的高级配置中的DDL同步配置为 “遇到DDL时任务报错停止” ，所以在识别到DDL事件时，会报此错误",
+			dynamicDescription = "Since the DDL synchronization in the advanced configuration of the source node of the task is configured as \"Stop Task Upon Encountering DDL\", the source node will report error to stop the task when it recognizes the DDL event",
+			dynamicDescriptionCN = "由于任务的源节点的高级配置中的DDL同步配置为 “遇到DDL时任务报错停止” ，所以源节点在识别到DDL事件时，任务会报错停止",
 			solution = "This error can be avoided by setting other DDL synchronization configurations, such as automatically ignoring all DDL events or synchronizing DDL events",
 			solutionCN = "可以通过设置其他DDL同步配置来避免此报错，如自动忽略所有DDL事件或同步DDL事件"
 	)
diff --git a/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/MilestoneAspectTask.java b/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/MilestoneAspectTask.java
index 599e15e7c2..0c7123f5d8 100644
--- a/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/MilestoneAspectTask.java
+++ b/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/MilestoneAspectTask.java
@@ -15,6 +15,7 @@
 import io.tapdata.aspect.task.AbstractAspectTask;
 import io.tapdata.aspect.task.AspectTaskSession;
 import io.tapdata.aspect.taskmilestones.*;
+import io.tapdata.exception.TapCodeException;
 import io.tapdata.exception.TmUnavailableException;
 import io.tapdata.milestone.constants.MilestoneStatus;
 import io.tapdata.milestone.entity.MilestoneEntity;
@@ -309,7 +310,11 @@ protected Void handleRetry(RetryLifeCycleAspect aspect) {
 //                aspect.getSuccess());
         // 1. get running milestone
         MilestoneEntity runningMilestone = milestones.values().stream().filter(m -> m.getStatus() == MilestoneStatus.RUNNING).findFirst().orElse(null);
-        // 2. update retry status to milestone
+        // 2. get cdc milestone when all finish
+        if (runningMilestone == null) {
+            runningMilestone = milestones.values().stream().filter(m -> KPI_CDC.equals(m.getCode())).findFirst().orElse(null);
+        }
+        // 3. update retry status to milestone
         Optional.ofNullable(runningMilestone).ifPresent(milestoneEntity -> {
             milestoneEntity.setRetrying(aspect.isRetrying());
             milestoneEntity.setRetryTimes(aspect.getRetryTimes());
@@ -502,6 +507,9 @@ protected <T extends AbsDataNodeErrorAspect<T>> void setError(T aspect, Mileston
         if (aspect.getError() != null && m.getStackMessage() == null) {
             m.setStackMessage(ExceptionUtils.getStackTrace(aspect.getError()));
         }
+        if (aspect.getError() instanceof TapCodeException) {
+            m.setDynamicDescriptionParameters(((TapCodeException)aspect.getError()).getDynamicDescriptionParameters());
+        }
     }
     protected <T extends EngineDeductionAspect> void setError(T aspect, MilestoneEntity m) {
         m.setEnd(System.currentTimeMillis());
@@ -513,6 +521,10 @@ protected <T extends EngineDeductionAspect> void setError(T aspect, MilestoneEnt
         if (aspect.getError() != null && m.getStackMessage() == null) {
             m.setStackMessage(ExceptionUtils.getStackTrace(aspect.getError()));
         }
+
+        if (aspect.getError() instanceof TapCodeException) {
+            m.setDynamicDescriptionParameters(((TapCodeException)aspect.getError()).getDynamicDescriptionParameters());
+        }
     }
 
     protected Consumer<MilestoneEntity> getErrorConsumer(String errorMessage) {
diff --git a/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/entity/MilestoneEntity.java b/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/entity/MilestoneEntity.java
index cb6fa8b743..3d5eaf99b1 100644
--- a/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/entity/MilestoneEntity.java
+++ b/iengine/modules/milestone-module/src/main/java/io/tapdata/milestone/entity/MilestoneEntity.java
@@ -20,6 +20,7 @@ public class MilestoneEntity implements Serializable {
     private String errorCode;
     private String errorMessage;
     private String stackMessage;
+    private String[] dynamicDescriptionParameters;
     private Long begin;
     private Long end;
     private Long progress;
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/BlankObsLogger.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/BlankObsLogger.java
index 09fb5fe301..399a7871ab 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/BlankObsLogger.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/BlankObsLogger.java
@@ -15,6 +15,11 @@ public MonitoringLogsDto.MonitoringLogsDtoBuilder logBaseBuilder() {
 		return null;
 	}
 
+	@Override
+	public void trace(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+
+	}
+
 	@Override
 	public void debug(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
 
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/LoggingAspectTask.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/LoggingAspectTask.java
index 389faf6e69..48ade0c13a 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/LoggingAspectTask.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/LoggingAspectTask.java
@@ -501,7 +501,9 @@ public Void handleCreateTableFuc(CreateTableFuncAspect aspect) {
 						aspect.getCreateTableEvent());
 				break;
 			case CreateTableFuncAspect.STATE_END:
-				if (null != aspect.getCreateTableOptions() && aspect.getCreateTableOptions().getTableExists()) {
+				if (null != aspect.getCreateTableOptions() &&
+						aspect.getCreateTableOptions().getTableExists() != null &&
+						aspect.getCreateTableOptions().getTableExists()) {
 					getObsLogger(node).info("The table {} has already exist.", aspect.getCreateTableEvent().getTable().getName());
 				}
 				break;
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/ObsLogger.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/ObsLogger.java
index e37a5ecc17..b9d58fe6f3 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/ObsLogger.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/ObsLogger.java
@@ -20,8 +20,12 @@ public abstract class ObsLogger implements Serializable, TapLogger {
 
 	private static final long serialVersionUID = -2297877575311105139L;
 
+	public static final String LOG_TAG_SOURCE_FOR_USER_SCRIPT = "src=user_script";
+
 	public abstract MonitoringLogsDto.MonitoringLogsDtoBuilder logBaseBuilder();
 
+	public abstract void trace(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params);
+
 	public abstract void debug(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params);
 
 	public abstract void info(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params);
@@ -44,6 +48,10 @@ public abstract class ObsLogger implements Serializable, TapLogger {
 
 	// debug level public logger api
 
+	public void trace(String message, Object... params) {
+		trace(this::logBaseBuilder, message, params);
+	}
+
 	public void debug(String message, Object... params) {
 		debug(this::logBaseBuilder, message, params);
 	}
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLogger.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLogger.java
index 3a21f44e6d..889938a7f3 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLogger.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLogger.java
@@ -49,8 +49,7 @@ public class TaskLogger extends ObsLogger {
 	private String taskRecordId;
 	@Getter
 	private String taskName;
-	private LogLevel level;
-	private LogLevel formerLevel;
+	private boolean enableDebugLogger;
 	private Long recordCeiling;
 	private Long intervalCeiling;
 	private boolean testTask;
@@ -80,7 +79,8 @@ protected TaskLogger(TaskDto taskDto, BiConsumer<String, LogLevel> consumer) {
 			// when preview task, no need to add any appender
 		} else {
 			this.witAppender(this.fileAppender(taskId))
-					.witAppender(this.obsHttpTMAppender(taskId));
+					.witAppender(this.obsHttpTMAppender(taskId))
+					.witAppender(this.debugFileAppender(taskId));
 		}
 
 		// add close debug consumer
@@ -95,6 +95,15 @@ private WithAppender<MonitoringLogsDto> fileAppender(String taskId){
 		};
 	}
 
+	private WithAppender<MonitoringLogsDto> debugFileAppender(String taskId){
+		return () -> {
+			// add file appender
+			String workDir = GlobalConstant.getInstance().getConfigurationCenter().getConfig(ConfigurationCenter.WORK_DIR).toString();
+			return (BaseTaskAppender<MonitoringLogsDto>) FileAppender.create(workDir, taskId + "_debug")
+					.include(LogLevel.DEBUG);
+		};
+	}
+
 	private WithAppender<MonitoringLogsDto> obsHttpTMAppender(String taskId){
 		return () -> {
 			// add tm appender
@@ -124,27 +133,16 @@ static TaskLogger create(TaskDto taskDto, BiConsumer<String, LogLevel> consumer)
 
 	TaskLogger withTaskLogSetting(String level, Long recordCeiling, Long intervalCeiling) {
 		LogLevel logLevel = LogLevel.getLogLevel(level);
-		if (this.level == logLevel) {
-			return this;
-		}
-		this.formerLevel = this.level;
-		this.level = logLevel;
-		if (this.level.isDebug()) {
-			if (null == recordCeiling && null == intervalCeiling) {
-				this.recordCeiling = RECORD_CEILING_DEFAULT;
-				this.intervalCeiling = System.currentTimeMillis() + INTERVAL_CEILING_DEFAULT * 1000;
-				return this;
-			}
 
-			this.recordCeiling = recordCeiling;
-			if (intervalCeiling != null) {
-				this.intervalCeiling = System.currentTimeMillis() + intervalCeiling * 1000;
-			}
+		if (logLevel.isDebug() && !enableDebugLogger) {
+			enableDebugLogger = true;
+
+			this.recordCeiling = recordCeiling == null ? RECORD_CEILING_DEFAULT : recordCeiling;
+			this.intervalCeiling = intervalCeiling == null ?
+					System.currentTimeMillis() + INTERVAL_CEILING_DEFAULT * 1000 :
+					System.currentTimeMillis() + intervalCeiling * 1000;
 			return this;
 		}
-
-		this.recordCeiling = null;
-		this.intervalCeiling = null;
 		return this;
 	}
 
@@ -153,30 +151,31 @@ private String formatMessage(String message, Object... params) {
 	}
 
 	public boolean noNeedLog(String level) {
-		if (!this.level.isDebug()) {
-			return !this.level.shouldLog(level);
+		LogLevel logLevel = Optional.ofNullable(LogLevel.getLogLevel(level)).orElse(LogLevel.TRACE);
+
+		if (!logLevel.isDebug()) {
+			return false;
 		}
 
+		if (!enableDebugLogger)
+			return true;
+
 		boolean noNeedLog = false;
 		if (null != recordCeiling) {
 			recordCeiling--;
-			noNeedLog = recordCeiling <= 0;
+			noNeedLog = recordCeiling < 0;
 		}
 
 		if (!noNeedLog && null != intervalCeiling) {
 			noNeedLog = intervalCeiling < System.currentTimeMillis();
 		}
 
-		if (noNeedLog && this.level.isDebug()) {
-			// fix NPE when task start as DEBUG level
-			if (null == formerLevel) {
-				formerLevel = LogLevel.INFO;
-			}
-			this.level = formerLevel;
+		if (noNeedLog) {
+			this.enableDebugLogger = false;
 			this.recordCeiling = null;
 			this.intervalCeiling = null;
 			if (null != closeDebugConsumer) {
-				closeDebugConsumer.accept(taskId, formerLevel);
+				closeDebugConsumer.accept(taskId, LogLevel.INFO);
 			}
 		}
 
@@ -191,6 +190,18 @@ private MonitoringLogsDto.MonitoringLogsDtoBuilder call(Callable<MonitoringLogsD
 		}
 	}
 
+	public void trace(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+		if (noNeedLog(LogLevel.TRACE.getLevel())) {
+			return;
+		}
+
+		MonitoringLogsDto.MonitoringLogsDtoBuilder builder = call(callable);
+		builder.level(Level.TRACE.toString());
+		builder.message(formatMessage(message, params));
+
+		logAppendFactory.appendLog(builder.build());
+	}
+
 	public void debug(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
 		if (noNeedLog(LogLevel.DEBUG.getLevel())) {
 			return;
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLoggerNodeProxy.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLoggerNodeProxy.java
index dfa35e3a77..da265dfcab 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLoggerNodeProxy.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/TaskLoggerNodeProxy.java
@@ -33,6 +33,10 @@ TaskLoggerNodeProxy withTags(List<String> tags) {
 		return this;
 	}
 
+	public void trace(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+		taskLogger.trace(callable, message, params);
+	}
+
 	public void debug(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
 		taskLogger.debug(callable, message, params);
 	}
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/AppenderFactory.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/AppenderFactory.java
index 283f4ac030..0455b891e6 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/AppenderFactory.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/AppenderFactory.java
@@ -23,15 +23,10 @@
 import java.io.File;
 import java.io.Serializable;
 import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 /**
  * @author jackin
@@ -39,6 +34,7 @@
  **/
 public class AppenderFactory implements Serializable {
 	public static final String FILE_APPENDER_TAILER_ID = "FILE_APPENDER_TAILER";
+	public static final String DEBUG_FILE_APPENDER_TAILER_ID = "DEBUG_FILE_APPENDER_TAILER";
 	public static final String TM_APPENDER_TAILER_ID= "TM_APPENDER_TAILER";
 	private volatile static AppenderFactory INSTANCE;
 
@@ -110,7 +106,12 @@ protected void readMessageFromCacheQueue(ExcerptTailer tailer,String tailerType)
 	protected void appenderAppendLog(MonitoringLogsDto.MonitoringLogsDtoBuilder builder, String tailerType) {
 		MonitoringLogsDto monitoringLogsDto = builder.build();
 		String taskId = monitoringLogsDto.getTaskId();
-		List<Appender<MonitoringLogsDto>> appenders = appenderMap.get(taskId);
+		List<Appender<MonitoringLogsDto>> appenders = appenderMap.keySet()
+				.stream().filter(k -> k.startsWith(taskId))
+				.map(appenderMap::get)
+				.filter(Objects::nonNull)
+				.flatMap(Collection::stream)
+				.collect(Collectors.toList());
 		if (CollectionUtils.isNotEmpty(appenders)) {
 			appenders.stream().filter(appender -> null != appender).filter((appender -> {
 				if (FILE_APPENDER_TAILER_ID.equals(tailerType)) {
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/FileAppender.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/FileAppender.java
index e7c9026fba..86112d29aa 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/FileAppender.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/FileAppender.java
@@ -2,9 +2,11 @@
 
 import com.tapdata.tm.commons.schema.MonitoringLogsDto;
 import io.tapdata.log.CustomPatternLayout;
+import io.tapdata.observable.logging.LogLevel;
 import io.tapdata.observable.logging.ObsLoggerFactory;
 import io.tapdata.observable.logging.util.Conf.LogConfiguration;
 import io.tapdata.observable.logging.util.LogUtil;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
@@ -22,7 +24,7 @@
 import org.apache.logging.log4j.core.layout.PatternLayout;
 
 import java.io.File;
-import java.util.List;
+import java.util.*;
 
 /**
  * @author jackin
@@ -36,6 +38,7 @@ public class FileAppender extends BaseTaskAppender<MonitoringLogsDto> {
 	private final Logger logger;
 	private final String workDir;
 	private String logsPath;
+	private Set<String> includeLogLevel;
 
 	public RollingFileAppender getRollingFileAppender() {
 		return rollingFileAppender;
@@ -47,7 +50,7 @@ private FileAppender(String workDir, String taskId) {
 		super(taskId);
 		this.workDir = workDir;
 		this.logger = LogManager.getLogger(LOGGER_NAME_PREFIX + taskId);
-		Configurator.setLevel(LOGGER_NAME_PREFIX, Level.DEBUG);
+		Configurator.setLevel(LOGGER_NAME_PREFIX + taskId, Level.TRACE);
 	}
 
 	public String getLogsPath() {
@@ -60,6 +63,8 @@ public static FileAppender create(String workDir, String taskId) {
 	@Override
 	public void append(MonitoringLogsDto log) {
 		final String level = log.getLevel();
+		if (CollectionUtils.isNotEmpty(includeLogLevel) && !includeLogLevel.contains(level))
+			return;
 		switch (level) {
 			case "DEBUG":
 				logger.debug(log.formatMonitoringLogMessage());
@@ -76,6 +81,9 @@ public void append(MonitoringLogsDto log) {
 			case "FATAL":
 				logger.fatal(log.formatMonitoringLogMessage());
 				break;
+			case "TRACE":
+				logger.trace(log.formatMonitoringLogMessage());
+				break;
 		}
 	}
 
@@ -135,4 +143,15 @@ public void stop() {
 			removeAppenders((org.apache.logging.log4j.core.Logger) logger);
 		}
 	}
+
+	public FileAppender include(LogLevel... level) {
+		if (level.length == 0) {
+			return this;
+		}
+		if (includeLogLevel == null) {
+			includeLogLevel = new HashSet<>();
+		}
+		Arrays.stream(level).map(LogLevel::getLevel).forEach(includeLogLevel::add);
+		return this;
+	}
 }
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/ObsHttpTMLog4jAppender.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/ObsHttpTMLog4jAppender.java
index 2f5992265f..d7bf838278 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/ObsHttpTMLog4jAppender.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/logging/appender/ObsHttpTMLog4jAppender.java
@@ -22,6 +22,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
+import static io.tapdata.observable.logging.ObsLogger.LOG_TAG_SOURCE_FOR_USER_SCRIPT;
+
 /**
  * @author samuel
  * @Description
@@ -85,7 +87,7 @@ private void consumeAndInsertLogs() {
 
 	protected void callTmApiInsertLogs(List<String> bufferList) {
 		List<String> tmp = bufferList.stream()
-				.filter(r -> !r.contains("src=user_script") || TokenBucketRateLimiter.get().tryAcquire(taskId))
+				.filter(r -> !r.contains(LOG_TAG_SOURCE_FOR_USER_SCRIPT) || TokenBucketRateLimiter.get().tryAcquire(taskId))
 				.collect(Collectors.toList());
 
         if (CollectionUtils.isNotEmpty(tmp)) {
diff --git a/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/SupervisorAspectTask.java b/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/SupervisorAspectTask.java
index c268c542d7..ead301727a 100644
--- a/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/SupervisorAspectTask.java
+++ b/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/SupervisorAspectTask.java
@@ -19,7 +19,7 @@
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
-@AspectTaskSession(includeTypes = {TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC}, ignoreErrors = false, order = 1)
+@AspectTaskSession(includeTypes = {TaskDto.SYNC_TYPE_MIGRATE, TaskDto.SYNC_TYPE_SYNC, TaskDto.SYNC_TYPE_LOG_COLLECTOR}, ignoreErrors = false, order = 1)
 public class SupervisorAspectTask extends AbstractAspectTask {
     private static final String TAG = SupervisorAspectTask.class.getSimpleName();
     @Bean
diff --git a/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/TaskResourceSupervisorManager.java b/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/TaskResourceSupervisorManager.java
index 554b755389..401cb4ebfd 100644
--- a/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/TaskResourceSupervisorManager.java
+++ b/iengine/modules/task-resource-supervisor-module/src/main/java/io/tapdata/supervisor/TaskResourceSupervisorManager.java
@@ -28,14 +28,14 @@ public class TaskResourceSupervisorManager implements MemoryFetcher {
     private String userId;
     private String processId;
     private Long lastCleanThreadGroupTime = null;
-    public final Long CHECK_THREAD_GROUP_POOL_PERIOD = 10 * 60L;
+    public static final Long CHECK_THREAD_GROUP_POOL_PERIOD = 10 * 60L;
 
-    public final Long CHECK_THREAD_GROUP_POOL_INITIAL_DELAY = 30L;
+    public static final Long CHECK_THREAD_GROUP_POOL_INITIAL_DELAY = 30L;
 
 
-    public String CLEAN_LEAKED_THREAD_GROUP_MINUTES = System.getenv().getOrDefault("CLEAN_LEAKED_THREAD_GROUP_MINUTES", "30");
+    public static final String CLEAN_LEAKED_THREAD_GROUP_MINUTES = System.getenv().getOrDefault("CLEAN_LEAKED_THREAD_GROUP_MINUTES", "30");
 
-    public String CLEAN_LEAKED_THREAD_GROUP_THRESHOLD = System.getenv().getOrDefault("CLEAN_LEAKED_THREAD_GROUP_THRESHOLD", "1000");
+    public static final String CLEAN_LEAKED_THREAD_GROUP_THRESHOLD = System.getenv().getOrDefault("CLEAN_LEAKED_THREAD_GROUP_THRESHOLD", "1000");
 
 
     public void cleanThreadGroup() {
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2.java b/manager/tm-api/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2.java
index d4218bbc1a..e1af81ab71 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2.java
@@ -44,4 +44,5 @@ public interface MeasurementServiceV2 {
 
     void queryTableMeasurement(String taskId, TableStatusInfoDto tableStatusInfoDto);
 
+    void cleanRemovedTableMeasurement(String taskId, String taskRecordId, String tableName);
 }
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/monitoringlogs/param/MonitoringLogQueryParam.java b/manager/tm-api/src/main/java/com/tapdata/tm/monitoringlogs/param/MonitoringLogQueryParam.java
index e686f022fa..937e0a93c3 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/monitoringlogs/param/MonitoringLogQueryParam.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/monitoringlogs/param/MonitoringLogQueryParam.java
@@ -31,6 +31,8 @@ public class MonitoringLogQueryParam {
     private String taskRecordId;
     private String search;
     private List<String> levels;
+    private List<String> includeLogTags;
+    private List<String> excludeLogTags;
 
     private String type = "monitor";
 
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/user/service/UserService.java b/manager/tm-api/src/main/java/com/tapdata/tm/user/service/UserService.java
index 829ef8ef78..6dfb304641 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/user/service/UserService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/user/service/UserService.java
@@ -68,5 +68,7 @@ public UserService(@NonNull UserRepository repository) {
     public abstract TestResponseDto testLoginByLdap(TestLdapDto testAdDto);
 
     public abstract boolean loginByLdap(String username, String password);
+
+    public abstract UserDto getUserDetail(String userId);
 }
 
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/error/TapDynamicTableNameExCode_35.java b/manager/tm-common/src/main/java/com/tapdata/tm/error/TapDynamicTableNameExCode_35.java
index f0af22fd24..29b86116e8 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/error/TapDynamicTableNameExCode_35.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/error/TapDynamicTableNameExCode_35.java
@@ -13,12 +13,12 @@
 public interface TapDynamicTableNameExCode_35 {
     @TapExCode
     String UNKNOWN_ERROR = "35001";
-
+    //TODO
     @TapExCode(
-            describe = "This error occurs when a dynamic date suffix cannot be created to generate a dynamic table name object when the target table of the development task is enabled",
-            describeCN = "当开发任务的目标节点开启动态日期后缀时，当无法创建动态生成表名规则对象时，会出现此错误",
+            describe = "When the dynamic date suffix is enabled on the target node of the development task, the dynamic generation table name rule object cannot be created",
+            describeCN = "当开发任务的目标节点开启动态日期后缀时，无法创建动态生成表名规则对象",
             dynamicDescription = "Failed to generate an object for a dynamic table name whose creation rule is {}",
-            dynamicDescriptionCN = "创建规则为{} 的动态表名生成对象时失败",
+            dynamicDescriptionCN = "生成动态表名规则失败，规则：{}",
             type = TapExType.PARAMETER_INVALID,
             level = TapExLevel.NORMAL
     )
@@ -26,7 +26,7 @@ public interface TapDynamicTableNameExCode_35 {
 
     @TapExCode(
             describe = "Fail to get invocation for dynamic table name",
-            describeCN = "当开发任务的目标节点开启动态日期后缀时，当创建动态生成表名规则对象时发生报错，则会抛出此错误",
+            describeCN = "当开发任务的目标节点开启动态日期后缀时，创建动态生成表名规则对象时发生报错",
             type = TapExType.PARAMETER_INVALID,
             level = TapExLevel.NORMAL
     )
@@ -34,7 +34,7 @@ public interface TapDynamicTableNameExCode_35 {
 
     @TapExCode(
             describe = "Dynamic method not find of Dynamic table name generation rule",
-            describeCN = "当开发任务目标节点开启动态日期后缀时，找不到动态表名生成规则的生成方法时，则会报此错误",
+            describeCN = "当开发任务目标节点开启动态日期后缀时，找不到动态表名生成规则的生成方法",
             type = TapExType.PARAMETER_INVALID,
             level = TapExLevel.NORMAL
     )
@@ -42,7 +42,7 @@ public interface TapDynamicTableNameExCode_35 {
 
     @TapExCode(
             describe = "Can not access dynamic table name generation method",
-            describeCN = "当开发任务目标节点开启动态日期后缀时，无法调用动态表名生成规则的生成方法时，则会报此错误",
+            describeCN = "当开发任务目标节点开启动态日期后缀时，无法调用动态表名生成规则的生成方法时",
             solution = "Please select a supported dynamic table name generation rule",
             solutionCN = "请选择一个已支持的动态表名生成规则",
             type = TapExType.PARAMETER_INVALID,
diff --git a/manager/tm/pom.xml b/manager/tm/pom.xml
index c128fdf7b2..078ee2716c 100644
--- a/manager/tm/pom.xml
+++ b/manager/tm/pom.xml
@@ -522,6 +522,10 @@
 			<type>pom</type>
 			<scope>import</scope>
 		</dependency>
+		<dependency>
+			<groupId>com.github.ben-manes.caffeine</groupId>
+			<artifactId>caffeine</artifactId>
+		</dependency>
 
 		<dependency>
 			<groupId>org.mockito</groupId>
diff --git a/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
index a36d168aff..3b870dcb10 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
@@ -31,6 +31,8 @@
 import org.apache.commons.lang3.StringUtils;
 import org.bson.Document;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.cache.caffeine.CaffeineCacheManager;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -49,6 +51,9 @@ public class SettingsServiceImpl implements SettingsService {
 
     private AlarmMailService alarmMailService;
 
+    @Qualifier("caffeineCache")
+    private CaffeineCacheManager caffeineCacheManager;
+
     /**
      * 有value 则返回value, 没有则返回default_value
      *
@@ -191,7 +196,11 @@ public List<SettingsDto> findALl(String decode, Filter filter) {
             settingsList.add(settings);
         } else {
             if(isCloud()){
-                settingsList = findAll(Query.query(Criteria.where("key").in("buildProfile","threshold","logLevel","job_cdc_record")));
+                Query query = Query.query(Criteria.where("key").in("buildProfile", "threshold", "logLevel", "job_cdc_record"));
+                settingsList = Optional.ofNullable(caffeineCacheManager.getCache("cloudSettings"))
+                        .map(cache -> cache.get("cloudSettings", () -> findAll(query)))
+                        .orElseGet(() -> findAll(query));
+
             }else {
                 settingsList = settingsRepository.findAll();
             }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/config/CacheConfig.java b/manager/tm/src/main/java/com/tapdata/tm/config/CacheConfig.java
index aa038de3d4..0fb62fb314 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/config/CacheConfig.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/config/CacheConfig.java
@@ -1,10 +1,16 @@
 package com.tapdata.tm.config;
 
+import com.github.benmanes.caffeine.cache.Caffeine;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cache.CacheManager;
 import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.cache.caffeine.CaffeineCacheManager;
 import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author lg<lirufei0808 @ gmail.com>
@@ -18,5 +24,24 @@ public class CacheConfig {
     public CacheManager memoryCacheManager() {
         return new ConcurrentMapCacheManager();
     }
+    @Value("${cache.expire:10}")
+    private Integer expire;
+    @Value("${cache.initialCapacity:100}")
+    private Integer initialCapacity;
+    @Value("${cache.maximumSize:500}")
+    private Integer maximumSize;
+
+    @Bean("caffeineCache")
+    @Primary
+    public CaffeineCacheManager cacheManager() {
+        CaffeineCacheManager cacheManager = new CaffeineCacheManager();
+        cacheManager.setCaffeine(Caffeine.newBuilder()
+                .expireAfterWrite(expire, TimeUnit.MINUTES)
+                .initialCapacity(initialCapacity)
+                .maximumSize(maximumSize)
+        );
+        return cacheManager;
+    }
+
 
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java b/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
index 173824c8af..059eb2f822 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
@@ -79,6 +79,11 @@
 public class MeasurementServiceV2Impl implements MeasurementServiceV2 {
     public static final String REPLICATE_LAG = "replicateLag";
     public static final String TASK_ID = "taskId";
+    public static final String TAGS_TASK_ID = "tags.taskId";
+    public static final String TAGS_TASK_RECORD_ID = "tags.taskRecordId";
+    public static final String TAGS_TYPE = "tags.type";
+    public static final String TABLE = "table";
+    public static final String TAGS_TABLE = "tags.table";
     private final MongoTemplate mongoOperations;
     private final MetadataInstancesService metadataInstancesService;
     private final TaskService taskService;
@@ -113,7 +118,7 @@ private void addBulkAgentMeasurement(List<SampleRequest> sampleRequestList) {
             }
 
             Date theDate = TimeUtil.cleanTimeAfterMinute(date);
-            if (!"table".equals(tags.get("type"))) {
+            if (!TABLE.equals(tags.get("type"))) {
                 criteria.and(MeasurementEntity.FIELD_DATE).is(theDate);
             }
 
@@ -136,7 +141,7 @@ private void addBulkAgentMeasurement(List<SampleRequest> sampleRequestList) {
             Update update = new Update()
                     .min(MeasurementEntity.FIELD_FIRST, requestSample.get().getDate())
                     .max(MeasurementEntity.FIELD_LAST, requestSample.get().getDate());
-            if ("table".equals(tags.get("type"))) {
+            if (TABLE.equals(tags.get("type"))) {
                 update.set(MeasurementEntity.FIELD_SAMPLES, Collections.singletonList(sampleMap));
                 update.set(MeasurementEntity.FIELD_DATE, theDate);
             } else {
@@ -308,7 +313,7 @@ protected Map<String, Sample> getInstantSamples(MeasurementQueryParam.Measuremen
             String format = String.format(TAG_FORMAT, entry.getKey());
             String value = entry.getValue();
             criteria.and(format).is(value);
-            if (format.equals("tags.type")) {
+            if (format.equals(TAGS_TYPE)) {
                 if ("task".equals(value)) {
                     typeIsTask = true;
                 } else if ("engine".equals(value)) {
@@ -317,10 +322,10 @@ protected Map<String, Sample> getInstantSamples(MeasurementQueryParam.Measuremen
                     typeIsNode = true;
                 }
             }
-            if (format.equals("tags.taskId")) {
+            if (format.equals(TAGS_TASK_ID)) {
                 taskId = value;
             }
-            if (format.equals("tags.taskRecordId")) {
+            if (format.equals(TAGS_TASK_RECORD_ID)) {
                 taskRecordId = value;
             }
         }
@@ -871,7 +876,7 @@ public void deleteTaskMeasurement(String taskId) {
         if (StringUtils.isEmpty(taskId)) {
             return;
         }
-        Query query = Query.query(Criteria.where("tags.taskId").is(taskId));
+        Query query = Query.query(Criteria.where(TAGS_TASK_ID).is(taskId));
         DeleteResult result = mongoOperations.remove(query, MeasurementEntity.class, MeasurementEntity.COLLECTION_NAME);
 
         log.info(" taskId :{}  删除了 {} 条记录", taskId, JsonUtil.toJson(result));
@@ -879,9 +884,9 @@ public void deleteTaskMeasurement(String taskId) {
 
     @Override
     public Long[] countEventByTaskRecord(String taskId, String taskRecordId) {
-        Query query = new Query(Criteria.where("tags.taskId").is(taskId)
-                .and("tags.taskRecordId").is(taskRecordId)
-                .and("tags.type").is("task")
+        Query query = new Query(Criteria.where(TAGS_TASK_ID).is(taskId)
+                .and(TAGS_TASK_RECORD_ID).is(taskRecordId)
+                .and(TAGS_TYPE).is("task")
                 .and(MeasurementEntity.FIELD_GRANULARITY).is(Granularity.GRANULARITY_MINUTE)
                 .and(MeasurementEntity.FIELD_DATE).lte(new Date()));
         query.with(Sort.by(MeasurementEntity.FIELD_DATE).descending());
@@ -927,19 +932,19 @@ public List<String> findRunTable(String taskId, String taskRecordId) {
             return runTables;
         }
 
-        Criteria criteria = Criteria.where("tags.taskId").is(taskId)
-                .and("tags.taskRecordId").is(taskRecordId)
-                .and("tags.type").is("table")
+        Criteria criteria = Criteria.where(TAGS_TASK_ID).is(taskId)
+                .and(TAGS_TASK_RECORD_ID).is(taskRecordId)
+                .and(TAGS_TYPE).is(TABLE)
                 .and(MeasurementEntity.FIELD_GRANULARITY).is(Granularity.GRANULARITY_MINUTE);
 
         Query query = new Query(criteria);
-        query.fields().include("tags.table");
+        query.fields().include(TAGS_TABLE);
         List<MeasurementEntity> measurementEntities = mongoOperations.find(query, MeasurementEntity.class, MeasurementEntity.COLLECTION_NAME);
         if (CollectionUtils.isNotEmpty(measurementEntities)) {
             for (MeasurementEntity measurementEntity : measurementEntities) {
                 Map<String, String> tags = measurementEntity.getTags();
-                if (tags != null && tags.get("table") != null) {
-                    runTables.add(tags.get("table"));
+                if (tags != null && tags.get(TABLE) != null) {
+                    runTables.add(tags.get(TABLE));
                 }
             }
         }
@@ -962,9 +967,9 @@ public Page<TableSyncStaticVo> querySyncStatic(TableSyncStaticDto dto, UserDetai
             hasTableRenameNode = taskDto.getDag().getNodes().stream().anyMatch(n -> n instanceof TableRenameProcessNode);
         }
 
-        Criteria criteria = Criteria.where("tags.taskId").is(taskDto.getId().toHexString())
-                .and("tags.taskRecordId").is(taskRecordId)
-                .and("tags.type").is("table")
+        Criteria criteria = Criteria.where(TAGS_TASK_ID).is(taskDto.getId().toHexString())
+                .and(TAGS_TASK_RECORD_ID).is(taskRecordId)
+                .and(TAGS_TYPE).is(TABLE)
                 .and(MeasurementEntity.FIELD_GRANULARITY).is(Granularity.GRANULARITY_MINUTE);
 
         querySyncByTableName(dto,criteria);
@@ -1018,7 +1023,7 @@ public Page<TableSyncStaticVo> querySyncStatic(TableSyncStaticDto dto, UserDetai
 
         List<TableSyncStaticVo> result = new ArrayList<>();
         for (MeasurementEntity measurementEntity : measurementEntities) {
-            String originTable = measurementEntity.getTags().get("table");
+            String originTable = measurementEntity.getTags().get(TABLE);
             AtomicReference<String> originTableName = new AtomicReference<>();
             boolean finalHasTableRenameNode = hasTableRenameNode;
             List<TableNode> finalCollect = collect;
@@ -1099,16 +1104,16 @@ protected void fixSyncRate(List<TableSyncStaticVo> tableSyncStaticVos, TaskDto t
 
     public void querySyncByTableName(TableSyncStaticDto dto, Criteria criteria){
         if(StringUtils.isNotBlank(dto.getTableName())){
-            criteria.and("tags.table").regex(dto.getTableName());
+            criteria.and(TAGS_TABLE).regex(dto.getTableName());
         }
     }
 
 	@Override
     public List<SyncStatusStatisticsVo> queryTableSyncStatusStatistics(SyncStatusStatisticsParam param) {
 		AggregationResults<SyncStatusStatisticsVo> aggregationResults = mongoOperations.aggregate(Aggregation.newAggregation(
-			Aggregation.match(Criteria.where("tags.taskId").is(param.getTaskId())
-				.and("tags.taskRecordId").is(param.getTaskRecordId())
-				.and("tags.type").is("table")
+			Aggregation.match(Criteria.where(TAGS_TASK_ID).is(param.getTaskId())
+				.and(TAGS_TASK_RECORD_ID).is(param.getTaskRecordId())
+				.and(TAGS_TYPE).is(TABLE)
 				.and(MeasurementEntity.FIELD_GRANULARITY).is(Granularity.GRANULARITY_MINUTE)),
 			Aggregation.unwind("$ss"),
 			Aggregation.project(Fields.from(
@@ -1131,9 +1136,9 @@ public List<SyncStatusStatisticsVo> queryTableSyncStatusStatistics(SyncStatusSta
      */
     @Override
     public MeasurementEntity findLastMinuteByTaskId(String taskId) {
-        Criteria criteria = Criteria.where("tags.taskId").is(taskId)
+        Criteria criteria = Criteria.where(TAGS_TASK_ID).is(taskId)
                 .and("grnty").is("minute")
-                .and("tags.type").is("task");
+                .and(TAGS_TYPE).is("task");
 
         Query query = new Query(criteria);
         query.fields().include("ss", "tags");
@@ -1144,9 +1149,9 @@ public MeasurementEntity findLastMinuteByTaskId(String taskId) {
 
     @Override
     public void queryTableMeasurement(String taskId, TableStatusInfoDto tableStatusInfoDto) {
-        Criteria criteria = Criteria.where("tags.taskId").is(taskId)
+        Criteria criteria = Criteria.where(TAGS_TASK_ID).is(taskId)
                 .and("grnty").is("minute")
-                .and("tags.type").is("task");
+                .and(TAGS_TYPE).is("task");
         Query query = new Query(criteria);
         query.fields().include("ss", "tags");
         query.with(Sort.by("last").descending());
@@ -1173,4 +1178,16 @@ public void queryTableMeasurement(String taskId, TableStatusInfoDto tableStatusI
         }
 
     }
+
+    @Override
+    public void cleanRemovedTableMeasurement(String taskId, String taskRecordId, String tableName) {
+        Criteria criteria = Criteria.where(TAGS_TASK_ID).is(taskId)
+                .and(TAGS_TASK_RECORD_ID).is(taskRecordId)
+                .and(TAGS_TYPE).is(TABLE)
+                .and(TAGS_TABLE).is(tableName)
+                .and(MeasurementEntity.FIELD_GRANULARITY).is(Granularity.GRANULARITY_MINUTE);
+
+        Query query = new Query(criteria);
+        mongoOperations.remove(query, MeasurementEntity.class, MeasurementEntity.COLLECTION_NAME);
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImpl.java
index a0ed3b010f..d2d65802b7 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImpl.java
@@ -161,6 +161,13 @@ public Page<MonitoringLogsDto> query(MonitoringLogQueryParam param) {
         }
         criteria.and("level").in(levels);
 
+        if (CollectionUtils.isNotEmpty(param.getIncludeLogTags())) {
+            criteria.and("logTags").in(param.getIncludeLogTags());
+        }
+        if (CollectionUtils.isNotEmpty(param.getExcludeLogTags())) {
+            criteria.and("logTags").nin(param.getExcludeLogTags());
+        }
+
         // sort and _id filter, avoid duplicate data
         Sort sort = Sort.by("date");
         switch (param.getOrder()) {
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 d8729acdc2..f2a7b4720a 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
@@ -25,15 +25,11 @@
 import com.tapdata.tm.base.dto.ResponseMessage;
 import com.tapdata.tm.base.dto.TmPageable;
 import com.tapdata.tm.base.dto.Where;
-import com.tapdata.tm.commons.dag.logCollector.LogCollecotrConnConfig;
-import com.tapdata.tm.commons.dag.logCollector.LogCollectorNode;
 import com.tapdata.tm.ds.entity.DataSourceEntity;
 import com.tapdata.tm.monitor.service.BatchService;
 import com.tapdata.tm.shareCdcTableMapping.service.ShareCdcTableMappingService;
 import com.tapdata.tm.task.bean.*;
 import com.tapdata.tm.task.vo.*;
-import io.tapdata.pdk.apis.entity.Capability;
-import io.tapdata.pdk.apis.entity.TestItem;
 import io.tapdata.pdk.core.api.PDKIntegration;
 import org.apache.commons.io.FileUtils;
 import org.mockito.Mockito;
@@ -3973,6 +3969,7 @@ public void start(TaskDto taskDto, UserDetail user, String startFlag, boolean sy
      *                  第二位 是否开启打点任务      1 是   0 否
      */
     public void start(TaskDto taskDto, UserDetail user, String startFlag) {
+        cleanRemovedTableMeasurementIfNeed(taskDto);
         boolean canStart = iLicenseService.checkTaskPipelineLimit(taskDto, user);
         if (!canStart) throw new BizException("Task.LicenseScheduleLimit");
         if (TaskDto.TYPE_INITIAL_SYNC.equals(taskDto.getType()) && TaskDto.STATUS_COMPLETE.equals(taskDto.getStatus()) && !taskDto.getCrontabExpressionFlag()) {
@@ -4071,6 +4068,23 @@ public void start(TaskDto taskDto, UserDetail user, String startFlag) {
         }
     }
 
+    protected void cleanRemovedTableMeasurementIfNeed(TaskDto taskDto) {
+        DAG dag = taskDto.getDag();
+        if (null == dag || CollectionUtils.isEmpty(dag.getSourceNode()) || StringUtils.isBlank(taskDto.getTaskRecordId())) {
+            return;
+        }
+        DatabaseNode newFirst = dag.getSourceNode().getFirst();
+        if (newFirst.getTableNames() != null) {
+            List<String> runTables = measurementServiceV2.findRunTable(taskDto.getId().toHexString(), taskDto.getTaskRecordId());
+            List<String> tables = new ArrayList<>(newFirst.getTableNames());
+            runTables.forEach(tableName -> {
+                if (!tables.contains(tableName)) {
+                    measurementServiceV2.cleanRemovedTableMeasurement(taskDto.getId().toHexString(), taskDto.getTaskRecordId(), tableName);
+                }
+            });
+        }
+    }
+
     /**
      * @see com.tapdata.tm.statemachine.enums.DataFlowEvent#START
      * @param taskDto
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
index ec689ed086..3712f6acc6 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/LdpServiceImpl.java
@@ -128,6 +128,8 @@ public class LdpServiceImpl implements LdpService {
 	private static final String TASK_ID = "taskId";
 	private static final String SOURCE_TYPE = "sourceType";
 
+	public static final String TASK_TYPE = "type";
+
     @Override
 	@Lock(value = "user.userId", type = LockType.START_LDP_FDM, expireSeconds = 15)
 	public TaskDto createFdmTask(TaskDto task, boolean start, UserDetail user) {
@@ -529,8 +531,10 @@ public TaskDto createMdmTask(TaskDto task, String tagId, UserDetail user, boolea
 
 	@Override
 	public void afterLdpTask(String taskId, UserDetail user) {
-		TaskDto taskDto = taskService.findById(MongoUtils.toObjectId(taskId));
-		taskService.updateAfter(taskDto, user);
+		com.tapdata.tm.base.dto.Field exclusionQueryField = new com.tapdata.tm.base.dto.Field();
+		exclusionQueryField.put(TASK_TYPE, 0);
+		TaskDto taskDto = taskService.findById(MongoUtils.toObjectId(taskId), exclusionQueryField);
+        taskService.updateAfter(taskDto, user);
 		LiveDataPlatformDto platformDto = liveDataPlatformService.findOne(new Query(), user);
         if (platformDto == null) {
             return;
diff --git a/manager/tm/src/main/java/com/tapdata/tm/user/controller/UserController.java b/manager/tm/src/main/java/com/tapdata/tm/user/controller/UserController.java
index a1bdb21b29..fdc6909d5c 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/user/controller/UserController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/user/controller/UserController.java
@@ -41,6 +41,9 @@
 import org.apache.commons.lang3.StringUtils;
 import org.bson.types.ObjectId;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cache.caffeine.CaffeineCacheManager;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
@@ -90,6 +93,13 @@ public class UserController extends BaseController {
     @Autowired
     UserLogService userLogService;
 
+    @Value("#{'${spring.profiles.include:idaas}'.split(',')}")
+    private List<String> productList;
+    @Autowired
+    @Qualifier("caffeineCache")
+    private CaffeineCacheManager userCache;
+
+
     private static final String RC4_KEY = "Gotapd8";
 
     /**
@@ -182,30 +192,16 @@ public ResponseMessage<UserDto> getUserByToken() {
      */
     @GetMapping("{userId}")
     public ResponseMessage<UserDto> getUser(@PathVariable(value = "userId") String userId) {
-        UserDto userDto = userService.findById(toObjectId(userId));
-
-        //userDto.setLastUpdAt(userDto.getLastUpdAt());
-        userDto.setCreateTime(userDto.getCreateAt());
-        List<RoleMappingDto> roleMappingDtoList = roleMappingService.getUser(PrincipleType.USER, userId);
-        if (CollectionUtils.isNotEmpty(roleMappingDtoList)) {
-            List<ObjectId> objectIds = roleMappingDtoList.stream().map(RoleMappingDto::getRoleId).collect(Collectors.toList());
-            List<RoleDto> roleDtos = roleService.findAll(Query.query(Criteria.where("_id").in(objectIds)));
-            if (CollectionUtils.isNotEmpty(roleDtos)) {
-                roleDtos.forEach(roleDto -> roleMappingDtoList.stream()
-                        .filter(roleMappingDto -> roleDto.getId().toHexString().equals(roleMappingDto.getRoleId().toHexString()))
-                        .findFirst().ifPresent(roleMappingDto -> roleMappingDto.setRole(roleDto)));
-            }
-            userDto.setRoleMappings(roleMappingDtoList);
+        if (productList != null && productList.contains("dfs")) {
+            return success(Optional.ofNullable(userCache.getCache("userCache"))
+                    .map(cache -> cache.get(userId, () -> userService.getUserDetail(userId)))
+                    .orElseGet(() -> userService.getUserDetail(userId)));
         }
-
-        if (StringUtils.isNotBlank(userId)) {
-            userDto.setPermissions(permissionService.getCurrentPermission(userId));
-        }
-
-        return success(userDto);
+        return success(userService.getUserDetail(userId));
     }
 
 
+
     /**
      * 根据userID 查询用户权限
      *
diff --git a/manager/tm/src/main/java/com/tapdata/tm/user/service/UserServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/user/service/UserServiceImpl.java
index b2740ebcba..dfca867dcb 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/user/service/UserServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/user/service/UserServiceImpl.java
@@ -862,6 +862,25 @@ public boolean loginByLdap(String username, String password) {
         return false;
     }
 
+    @Override
+    public UserDto getUserDetail(String userId) {
+        UserDto userDto = findById(toObjectId(userId));
+        userDto.setCreateTime(userDto.getCreateAt());
+        List<RoleMappingDto> roleMappingDtoList = roleMappingService.getUser(PrincipleType.USER, userId);
+        if (CollectionUtils.isNotEmpty(roleMappingDtoList)) {
+            List<ObjectId> objectIds = roleMappingDtoList.stream().map(RoleMappingDto::getRoleId).collect(Collectors.toList());
+            List<RoleDto> roleDtos = roleService.findAll(Query.query(Criteria.where("_id").in(objectIds)));
+            if (CollectionUtils.isNotEmpty(roleDtos)) {
+                roleDtos.forEach(roleDto -> roleMappingDtoList.stream()
+                        .filter(roleMappingDto -> roleDto.getId().toHexString().equals(roleMappingDto.getRoleId().toHexString()))
+                        .findFirst().ifPresent(roleMappingDto -> roleMappingDto.setRole(roleDto)));
+            }
+            userDto.setRoleMappings(roleMappingDtoList);
+        }
+        userDto.setPermissions(permissionService.getCurrentPermission(userId));
+        return userDto;
+    }
+
     protected boolean searchUser(LdapLoginDto ldapLoginDto, String username) throws NamingException {
         String sAMAccountNameFilter = String.format("(sAMAccountName=%s)", username);
         String userPrincipalNameFilter = String.format("(userPrincipalName=%s)", username);
@@ -875,7 +894,10 @@ protected boolean searchUser(LdapLoginDto ldapLoginDto, String username) throws
             String[] searchBase = searchBases.split(";");
             for (String base : searchBase) {
                 boolean isExist = searchWithFilter(ctx, base, sAMAccountNameFilter, searchControls) || searchWithFilter(ctx, base, userPrincipalNameFilter, searchControls);
-                if (isExist) return true;
+                if (isExist) {
+                    ldapLoginDto.setBaseDN(base);
+                    return true;
+                }
             }
             return false;
         } catch (NamingException e) {
@@ -902,6 +924,7 @@ protected boolean searchWithFilter(DirContext ctx, String searchBase, String fil
     protected DirContext buildDirContext(LdapLoginDto ldapLoginDto) throws NamingException {
         String ldapUrl = ldapLoginDto.getLdapUrl();
         String bindDn = ldapLoginDto.getBindDN();
+        String baseDN = ldapLoginDto.getBaseDN();
         String password = ldapLoginDto.getPassword();
         Boolean ssl = ldapLoginDto.isSslEnable();
         String certFile = ldapLoginDto.getCert();
@@ -909,6 +932,10 @@ protected DirContext buildDirContext(LdapLoginDto ldapLoginDto) throws NamingExc
         env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");
         env.put(Context.PROVIDER_URL, ldapUrl);
         env.put(Context.SECURITY_AUTHENTICATION, "simple");
+        if (!bindDn.contains("@") && StringUtils.isNotBlank(baseDN)) {
+            String domain = convertBaseDnToDomain(baseDN);
+            bindDn = bindDn + "@" + domain;
+        }
         env.put(Context.SECURITY_PRINCIPAL, bindDn);
         env.put(Context.SECURITY_CREDENTIALS, password);
         if (ssl) {
@@ -923,6 +950,22 @@ protected DirContext buildDirContext(LdapLoginDto ldapLoginDto) throws NamingExc
         DirContext ctx = new InitialDirContext(env);
         return ctx;
     }
+
+    protected String convertBaseDnToDomain(String baseDn) {
+        String[] parts = baseDn.split(",");
+        StringBuilder domain = new StringBuilder();
+        for (String part : parts) {
+            part = part.trim();
+            if (part.startsWith("DC=") || part.startsWith("dc=")) {
+                if (domain.length() > 0) {
+                    domain.append(".");
+                }
+                domain.append(part.substring(3));
+            }
+        }
+        return domain.length() > 0 ? domain.toString() : null;
+    }
+
     protected SSLContext createSSLContext(InputStream certFile) throws Exception {
         // load custom cert
         CertificateFactory cf = CertificateFactory.getInstance("X.509");
diff --git a/manager/tm/src/main/resources/application-default.yml b/manager/tm/src/main/resources/application-default.yml
index 44545580ac..db519ce368 100644
--- a/manager/tm/src/main/resources/application-default.yml
+++ b/manager/tm/src/main/resources/application-default.yml
@@ -132,4 +132,10 @@ report:
     apiSecret: 'noI4OfyaSuCOaFbr7AKbxA'
     clientId: '796104786.1713955139'
   data:
-    oss: ${REPORT_DATA_OSS:false}
\ No newline at end of file
+    oss: ${REPORT_DATA_OSS:false}
+
+cache:
+  expire: 10 #缓存过期时间，单位分钟
+  initialCapacity: 100 #初始化缓存容量
+  maximumSize: 500 #最大缓存数
+
diff --git a/manager/tm/src/main/resources/init/dfs/3.18-1.json b/manager/tm/src/main/resources/init/dfs/3.18-1.json
new file mode 100644
index 0000000000..8929580bf8
--- /dev/null
+++ b/manager/tm/src/main/resources/init/dfs/3.18-1.json
@@ -0,0 +1,21 @@
+[
+  {
+    "update": "JavascriptFunctions",
+    "updates": [
+      {
+        "q": {
+          "className": "MapUtil",
+          "desc": "字典工具类"
+        },
+        "u": {
+          "$set": {
+            "className": "MapUtils",
+            "example": "//从给定的map中获取指定层级的值\nvar a = MapUtils.getValueByKey(map, 'a.b.c');"
+          }
+        },
+        "upsert": false,
+        "multi": true
+      }
+    ]
+  }
+]
diff --git a/manager/tm/src/main/resources/init/dfs/version b/manager/tm/src/main/resources/init/dfs/version
index 3572d23604..c53a1ce84d 100644
--- a/manager/tm/src/main/resources/init/dfs/version
+++ b/manager/tm/src/main/resources/init/dfs/version
@@ -1,1 +1,1 @@
-3.17-1
\ No newline at end of file
+3.18-1
\ No newline at end of file
diff --git a/manager/tm/src/main/resources/init/idaas/3.18-1.json b/manager/tm/src/main/resources/init/idaas/3.18-1.json
new file mode 100644
index 0000000000..8929580bf8
--- /dev/null
+++ b/manager/tm/src/main/resources/init/idaas/3.18-1.json
@@ -0,0 +1,21 @@
+[
+  {
+    "update": "JavascriptFunctions",
+    "updates": [
+      {
+        "q": {
+          "className": "MapUtil",
+          "desc": "字典工具类"
+        },
+        "u": {
+          "$set": {
+            "className": "MapUtils",
+            "example": "//从给定的map中获取指定层级的值\nvar a = MapUtils.getValueByKey(map, 'a.b.c');"
+          }
+        },
+        "upsert": false,
+        "multi": true
+      }
+    ]
+  }
+]
diff --git a/manager/tm/src/main/resources/init/idaas/version b/manager/tm/src/main/resources/init/idaas/version
index 3572d23604..c53a1ce84d 100644
--- a/manager/tm/src/main/resources/init/idaas/version
+++ b/manager/tm/src/main/resources/init/idaas/version
@@ -1,1 +1,1 @@
-3.17-1
\ No newline at end of file
+3.18-1
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 186674337c..80a7343471 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,7 +46,8 @@
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/script/ObsScriptLogger.java,
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/StopTaskOnErrorLog.java,
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/script/TapScriptLogger.java,
-            iengine/api/src/main/java/io/tapdata/aspect/TaskBatchSplitAspect.java
+            iengine/api/src/main/java/io/tapdata/aspect/TaskBatchSplitAspect.java,
+            manager/tm/src/main/java/com/tapdata/tm/config/*
         </sonar.coverage.exclusions>
         <sonar.exclusions>
             manager/**/dto/*,
@@ -75,7 +76,8 @@
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/script/ObsScriptLogger.java,
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/StopTaskOnErrorLog.java,
             iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/script/TapScriptLogger.java,
-            iengine/api/src/main/java/io/tapdata/aspect/TaskBatchSplitAspect.java
+            iengine/api/src/main/java/io/tapdata/aspect/TaskBatchSplitAspect.java,
+            manager/tm/src/main/java/com/tapdata/tm/config/*
         </sonar.exclusions>
         <sonar.test.exclusions>
             **/src/test/**


Test patch:
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 b346dc889e..33e66a2b51 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
@@ -22,7 +22,6 @@
 import com.tapdata.tm.commons.dag.process.UnwindProcessNode;
 import com.tapdata.tm.commons.schema.MetadataInstancesDto;
 import com.tapdata.tm.commons.task.dto.TaskDto;
-import com.tapdata.tm.commons.util.JsonUtil;
 import io.tapdata.aspect.CreateTableFuncAspect;
 import io.tapdata.aspect.DropTableFuncAspect;
 import io.tapdata.aspect.supervisor.DataNodeThreadGroupAspect;
@@ -52,7 +51,6 @@
 import io.tapdata.flow.engine.V2.node.hazelcast.data.pdk.concurrent.PartitionConcurrentProcessor;
 import io.tapdata.flow.engine.V2.node.hazelcast.data.pdk.concurrent.partitioner.Partitioner;
 import io.tapdata.flow.engine.V2.util.PdkUtil;
-import io.tapdata.flow.engine.V2.util.TapEventUtil;
 import io.tapdata.flow.engine.V2.util.SyncTypeEnum;
 import io.tapdata.metric.collector.ISyncMetricCollector;
 import io.tapdata.metric.collector.SyncMetricCollector;
@@ -70,7 +68,6 @@
 import io.tapdata.pdk.core.async.ThreadPoolExecutorEx;
 import io.tapdata.pdk.core.monitor.PDKInvocationMonitor;
 import io.tapdata.pdk.core.utils.CommonUtils;
-import io.tapdata.supervisor.TaskNodeInfo;
 import io.tapdata.supervisor.TaskResourceSupervisorManager;
 import io.tapdata.utils.UnitTestUtils;
 import lombok.SneakyThrows;
@@ -80,7 +77,6 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.internal.verification.Times;
-import org.springframework.beans.BeanUtils;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.time.Instant;
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 9f244bef0c..5810b826b0 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
@@ -11,8 +11,10 @@
 import com.tapdata.entity.TapdataHeartbeatEvent;
 import com.tapdata.entity.task.context.ProcessorBaseContext;
 import com.tapdata.tm.commons.dag.Edge;
+import io.github.openlg.graphlib.Graph;
 import io.tapdata.entity.schema.value.TapArrayValue;
 import io.tapdata.entity.schema.value.TapMapValue;
+import io.tapdata.flow.engine.V2.task.preview.node.HazelcastPreviewMergeNode;
 import io.tapdata.utils.AppType;
 import com.tapdata.entity.Connections;
 import com.tapdata.entity.SyncStage;
@@ -60,10 +62,12 @@
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.io.File;
+import java.lang.reflect.Field;
 import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
@@ -170,6 +174,30 @@ void testDoInitExternalStorageDaasAppType() {
 				assertNull(actualObj);
 			}
 		}
+
+		@Test
+		void testInitMergeTableProperties() {
+			when(dataProcessorContext.getNode()).thenReturn((Node) tableNode);
+			try {
+				hazelcastMergeNode.initMergeTableProperties();
+			} catch (TapCodeException e) {
+				assertEquals(TaskMergeProcessorExCode_16.WRONG_NODE_TYPE,e.getCode());
+			}
+		}
+
+	}
+
+	@Test
+	void testGetSourceTableNode() {
+		MergeTableNode mergeTableNode1 = mock(MergeTableNode.class);
+		when(dataProcessorContext.getNode()).thenReturn((Node)mergeTableNode1);
+		Graph graph = mock(Graph.class);
+		when(mergeTableNode1.getGraph()).thenReturn(graph);
+		try {
+			hazelcastMergeNode.getSourceTableNode("sourceId");
+		} catch (TapCodeException e) {
+			assertEquals(TaskMergeProcessorExCode_16.CANNOT_FOUND_PRE_NODE,e.getCode());
+		}
 	}
 
 	@Nested
@@ -187,10 +215,28 @@ void beforeEach() {
 			hazelcastMergeNode = spy(hazelcastMergeNode);
 			tapdataEvent = mock(TapdataEvent.class);
 			tapInsertRecordEvent = mock(TapInsertRecordEvent.class);
+			when(tapInsertRecordEvent.getTableId()).thenReturn("tableName");
+			List<String> nodeIds = new ArrayList<>();
+			nodeIds.add("sourceId");
+			when(tapdataEvent.getNodeIds()).thenReturn(nodeIds);
 			when(tapdataEvent.getTapEvent()).thenReturn(tapInsertRecordEvent);
 			constructIMap = mock(ConstructIMap.class);
 			mergeTableProperties = mock(MergeTableProperties.class);
 			sourcePkOrUniqueFieldMap = new HashMap<>();
+			TableNode tableNode = mock(TableNode.class);
+			when(tableNode.getTableName()).thenReturn("tableName");
+			when(tableNode.getName()).thenReturn("NodeName");
+			when(tableNode.getId()).thenReturn("sourceId");
+
+			List<Node> nodes=new ArrayList<>();
+			nodes.add(tableNode);
+
+			when(dataProcessorContext.getNodes()).thenReturn(nodes);
+			when(hazelcastMergeNode.getPreNode("sourceId")).thenReturn(nodes.get(0));
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("MongoEx");
+			ReflectionTestUtils.setField(hazelcastMergeNode,"externalStorageDto",externalStorageDto);
 			ReflectionTestUtils.setField(hazelcastMergeNode, "sourcePkOrUniqueFieldMap", sourcePkOrUniqueFieldMap);
 			ReflectionTestUtils.setField(hazelcastMergeNode, "mapIterator", new AllLayerMapIterator());
 		}
@@ -219,6 +265,58 @@ void testUpsertOneEventCacheMustCallDateTimeValue() {
 			hazelcastMergeNode.upsertCache(tapdataEvent, mergeTableProperties, constructIMap);
 			verify(hazelcastMergeNode, new Times(1)).transformDateTime(any(Map.class));
 		}
+		@Test
+		@SneakyThrows
+		@DisplayName("Upsert one event into cache must call transformDateTime method")
+		void testUpsertOneEventCacheThrowFindException() {
+			Map<String, Object> after = new HashMap<>();
+			after.put("id", 1);
+			after.put("create_time", new DateTime(Instant.now()));
+			when(tapInsertRecordEvent.getAfter()).thenReturn(after);
+			List<Map<String, String>> joinKeys = new ArrayList<Map<String, String>>() {{
+				add(new HashMap<String, String>() {{
+					put("source", "id");
+					put("target", "id");
+				}});
+			}};
+			when(mergeTableProperties.getJoinKeys()).thenReturn(joinKeys);
+			when(mergeTableProperties.getId()).thenReturn("sourceId");
+			sourcePkOrUniqueFieldMap.put("sourceId", new ArrayList<String>() {{
+				add("id");
+			}});
+			when(constructIMap.find(anyString())).thenThrow(new RuntimeException("connect Exception"));
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.upsertCache(tapdataEvent, mergeTableProperties, constructIMap);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.UPSERT_CACHE_FIND_BY_JOIN_KEY_FAILED,tapCodeException.getCode());
+		}
+		@Test
+		@SneakyThrows
+		@DisplayName("Upsert one event into cache must call transformDateTime method")
+		void testUpsertOneEventThrowUpsertException() {
+			Map<String, Object> after = new HashMap<>();
+			after.put("id", 1);
+			after.put("create_time", "");
+			when(tapInsertRecordEvent.getAfter()).thenReturn(after);
+			List<Map<String, String>> joinKeys = new ArrayList<Map<String, String>>() {{
+				add(new HashMap<String, String>() {{
+					put("source", "id");
+					put("target", "id");
+				}});
+			}};
+			when(mergeTableProperties.getJoinKeys()).thenReturn(joinKeys);
+			when(mergeTableProperties.getId()).thenReturn("sourceId");
+			sourcePkOrUniqueFieldMap.put("sourceId", new ArrayList<String>() {{
+				add("id");
+			}});
+			when(constructIMap.find(anyString())).thenReturn(null);
+			when(constructIMap.upsert(anyString(), any(Document.class))).thenThrow(new RuntimeException("throw Upsert Exception"));
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.upsertCache(tapdataEvent, mergeTableProperties, constructIMap);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.UPSERT_CACHE_FAILED,tapCodeException.getCode());
+
+		}
 
 		@Test
 		@SneakyThrows
@@ -254,6 +352,42 @@ void testUpsertManyEventCacheMustCallDateTimeValue() {
 			}}, mergeTableProperties, constructIMap);
 			verify(hazelcastMergeNode, new Times(2)).transformDateTime(any(Map.class));
 		}
+		@Test
+		@SneakyThrows
+		@DisplayName("Upsert many events into cache must call transformDateTime method")
+		void testUpsertManyEventWriteFailed() {
+			Map<String, Object> after = new HashMap<>();
+			after.put("id", 1);
+			after.put("create_time", new DateTime(Instant.now()));
+			when(tapInsertRecordEvent.getAfter()).thenReturn(after);
+			TapInsertRecordEvent tapInsertRecordEvent1 = mock(TapInsertRecordEvent.class);
+			when(tapInsertRecordEvent1.getAfter()).thenReturn(new HashMap<String, Object>() {{
+				put("id", 2);
+				put("create_time", new DateTime(Instant.now()));
+			}});
+			TapdataEvent tapdataEvent1 = mock(TapdataEvent.class);
+			when(tapdataEvent1.getTapEvent()).thenReturn(tapInsertRecordEvent1);
+			List<Map<String, String>> joinKeys = new ArrayList<Map<String, String>>() {{
+				add(new HashMap<String, String>() {{
+					put("source", "id");
+					put("target", "id");
+				}});
+			}};
+			when(mergeTableProperties.getJoinKeys()).thenReturn(joinKeys);
+			when(mergeTableProperties.getId()).thenReturn("sourceId");
+			sourcePkOrUniqueFieldMap.put("sourceId", new ArrayList<String>() {{
+				add("id");
+			}});
+			when(constructIMap.findAll(any(Set.class))).thenReturn(new HashMap<>());
+			when(constructIMap.insertMany(anyMap())).thenThrow(new RuntimeException("writeFailed"));
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.upsertCache(new ArrayList<TapdataEvent>() {{
+					add(tapdataEvent);
+					add(tapdataEvent1);
+				}}, mergeTableProperties, constructIMap);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.UPSERT_CACHES_FAILED,tapCodeException.getCode());
+		}
 
 		@Test
 		@SneakyThrows
@@ -2129,12 +2263,63 @@ void testNodeLoggerDebugEnabled() {
 			completableFuture.join();
 			verify(nodeLogger, timeout(1)).debug(anyString(), any(Object[].class));
 		}
+		@Test
+		void testLoopUpThrowException(){
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.lookup(tapdataEvent);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.LOOK_UP_MISSING_FROM_NODE_ID,tapCodeException.getCode());
+		}
 
 		@AfterEach
 		void tearDown() {
 			lookupThreadPool.shutdownNow();
 		}
 	}
+	@Nested
+	class PutInCheckJoinKeyUpdateCacheMapAndWriteSignTest{
+		HazelcastMergeNode mockHazelcastMergeNode;
+		Map<String,ConstructIMap<Document>> iMapMap;
+		String id="sourceId";
+		@BeforeEach
+		void init() {
+			mockHazelcastMergeNode=mock(HazelcastMergeNode.class);
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("Mongo External");
+			ObsLogger obsLogger = mock(ObsLogger.class);
+			Processor.Context context = mock(Processor.Context.class);
+			iMapMap=new HashMap<>();
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"jetContext",context);
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"obsLogger",obsLogger);
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"externalStorageDto",externalStorageDto);
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"checkJoinKeyUpdateCacheMap",iMapMap);
+			Node tableNode=mock(TableNode.class);
+			when(mockHazelcastMergeNode.getPreNode(id)).thenReturn(tableNode);
+			doCallRealMethod().when(mockHazelcastMergeNode).putInCheckJoinKeyUpdateCacheMapAndWriteSign(anyString(),anyString());
+			when(mockHazelcastMergeNode.copyExternalStorage(anyInt())).thenReturn(externalStorageDto);
+		}
+		@DisplayName("test putInCheckJoinKeyUpdateCacheMapAndWriteSign throw exception INIT_CHECK_UPDATE_JOIN_KEY_VALUE_CACHE_WRITE_SIGN_FAILED")
+		@SneakyThrows
+		@Test
+		void test1(){
+			ConstructIMap constructIMap = mock(ConstructIMap.class);
+			when(constructIMap.isEmpty()).thenReturn(true);
+			when(constructIMap.getName()).thenReturn("cacheName");
+			when(constructIMap.insert(anyString(),any())).thenThrow(new RuntimeException("insert Failed"));
+			try(MockedStatic<HazelcastMergeNode> hazelcastMergeNodeMockedStatic = mockStatic(HazelcastMergeNode.class);){
+				hazelcastMergeNodeMockedStatic.when(()->{HazelcastMergeNode.buildConstructIMap(any(),any(),any(),any());}).thenReturn(constructIMap);
+				when(mockHazelcastMergeNode.checkJoinKeyIncludePK(anyString())).thenReturn(null);
+				TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+					mockHazelcastMergeNode.putInCheckJoinKeyUpdateCacheMapAndWriteSign(id, "cacheName");
+				});
+				assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.INIT_CHECK_UPDATE_JOIN_KEY_VALUE_CACHE_WRITE_SIGN_FAILED);
+
+			}
+
+		}
+	}
 
 	@Nested
 	@DisplayName("Method doBatchCache test")
@@ -2168,6 +2353,233 @@ void testMainProcess() {
 			verify(hazelcastMergeNode, times(1)).cache(any(List.class));
 		}
 	}
+	@Nested
+	class TestDeleteCache{
+		HazelcastMergeNode mockHazelcastMergeNode;
+		ConstructIMap<Document> constructIMap;
+		MergeTableProperties mergeTableProperties;
+		@BeforeEach
+		void init(){
+			mockHazelcastMergeNode=mock(HazelcastMergeNode.class);
+			constructIMap=mock(ConstructIMap.class);
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("Mongo ExteralStorageDto");
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"externalStorageDto",externalStorageDto);
+			mergeTableProperties = mock(MergeTableProperties.class);
+			when(mergeTableProperties.getId()).thenReturn("nodeId");
+			Node tableNode=mock(TableNode.class);
+			when(tableNode.getName()).thenReturn("nodeName");
+			when(mockHazelcastMergeNode.getPreNode(anyString())).thenReturn(tableNode);
+		}
+		@DisplayName("test delete cache find error")
+		@Test
+		void test1() throws Exception {
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			Map<String,Object> before=new HashMap<>();
+			before.put("id",1);
+			before.put("joinKey",1);
+			TapDeleteRecordEvent deleteRecordEvent =TapDeleteRecordEvent.create();
+			deleteRecordEvent.setBefore(before);
+			tapdataEvent.setTapEvent(deleteRecordEvent);
+
+			String joinValue="1";
+			when(mockHazelcastMergeNode.getJoinValueKeyBySource(any(),any(),any())).thenReturn(joinValue);
+			String pkValue="1";
+			when(mockHazelcastMergeNode.getPkOrUniqueValueKey(anyMap(),any(MergeTableProperties.class),any())).thenReturn(pkValue);
+			when(constructIMap.find(anyString())).thenThrow(new RuntimeException("find Failed"));
+			doCallRealMethod().when(mockHazelcastMergeNode).encode(anyString());
+			doCallRealMethod().when(mockHazelcastMergeNode).getBefore(any());
+			doCallRealMethod().when(mockHazelcastMergeNode).deleteCache(any(),any(),any());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.deleteCache(tapdataEvent, mergeTableProperties, constructIMap);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.DELETE_CACHE_FIND_BY_JOIN_KEY_FAILED);
+		}
+		@DisplayName("test delete cache delete failed")
+		@Test
+		void test2() throws Exception {
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			Map<String,Object> before=new HashMap<>();
+			before.put("id",1);
+			before.put("joinKey",1);
+			TapDeleteRecordEvent deleteRecordEvent =TapDeleteRecordEvent.create();
+			deleteRecordEvent.setBefore(before);
+			tapdataEvent.setTapEvent(deleteRecordEvent);
+
+			String joinValue="1";
+			when(mockHazelcastMergeNode.getJoinValueKeyBySource(any(),any(),any())).thenReturn(joinValue);
+			String pkValue="1";
+			when(mockHazelcastMergeNode.getPkOrUniqueValueKey(anyMap(),any(MergeTableProperties.class),any())).thenReturn(pkValue);
+			Document document=new Document();
+			when(constructIMap.find(anyString())).thenReturn(document);
+			when(constructIMap.delete(anyString())).thenThrow(new RuntimeException("delete failed"));
+			doCallRealMethod().when(mockHazelcastMergeNode).encode(anyString());
+			doCallRealMethod().when(mockHazelcastMergeNode).getBefore(any());
+			doCallRealMethod().when(mockHazelcastMergeNode).deleteCache(any(),any(),any());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.deleteCache(tapdataEvent, mergeTableProperties, constructIMap);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.DELETE_CACHE_FAILED);
+		}
+	}
+	@Nested
+	class TestGetJoinValueKeyBySource{
+		HazelcastMergeNode mockHazelcastMergeNode;
+		ConstructIMap<Document> constructIMap;
+		MergeTableProperties mergeTableProperties;
+		@BeforeEach
+		void beforeInit(){
+			constructIMap=mock(ConstructIMap.class);
+			mockHazelcastMergeNode=mock(HazelcastMergeNode.class);
+			mergeTableProperties = mock(MergeTableProperties.class);
+			when(mergeTableProperties.getId()).thenReturn("nodeId");
+			Node tableNode=mock(TableNode.class);
+			when(tableNode.getName()).thenReturn("nodeName");
+			when(mockHazelcastMergeNode.getPreNode(anyString())).thenReturn(tableNode);
+		}
+		@DisplayName("test getJoinValueKeyBySource missing join key")
+		@Test
+		void test1(){
+			Map<String,Object> after=new HashMap<>();
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("Mongo ExteralStorageDto");
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"externalStorageDto",externalStorageDto);
+			List<Map<String, String>> joinKeys =new ArrayList<>();
+			when(mergeTableProperties.getJoinKeys()).thenReturn(joinKeys);
+			doCallRealMethod().when(mockHazelcastMergeNode).getJoinValueKeyBySource(after,mergeTableProperties,constructIMap);
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getJoinValueKeyBySource(after, mergeTableProperties, constructIMap);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.MISSING_SOURCE_JOIN_KEY_CONFIG);
+		}
+		@DisplayName("test getJoinValueKeyBySource after not have join key")
+		@Test
+		void test2(){
+			Map<String,Object> after=new HashMap<>();
+			after.put("id",1);
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("Mongo ExteralStorageDto");
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"externalStorageDto",externalStorageDto);
+			List<Map<String, String>> joinKeys =new ArrayList<>();
+			Map<String,String> sourceJoinKey=new HashMap<>();
+			sourceJoinKey.put(HazelcastPreviewMergeNode.JoinKeyEnum.SOURCE.getKey(), "joinKey");
+			joinKeys.add(sourceJoinKey);
+			when(mergeTableProperties.getJoinKeys()).thenReturn(joinKeys);
+			doCallRealMethod().when(mockHazelcastMergeNode).getJoinKeys(any(),any());
+			doCallRealMethod().when(mockHazelcastMergeNode).getJoinValueKeyBySource(after,mergeTableProperties,constructIMap);
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getJoinValueKeyBySource(after, mergeTableProperties, constructIMap);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.JOIN_KEY_VALUE_NOT_EXISTS);
+		}
+	}
+	@Nested
+	class TestGetPkOrUniqueValueKey{
+		HazelcastMergeNode mockHazelcastMergeNode;
+		ConstructIMap<Document> constructIMap;
+		MergeTableProperties mergeTableProperties;
+		Map<String, List<String>> sourcePkOrUniqueFieldMap;
+		@BeforeEach
+		void init(){
+			constructIMap=mock(ConstructIMap.class);
+			mockHazelcastMergeNode=mock(HazelcastMergeNode.class);
+			mergeTableProperties = mock(MergeTableProperties.class);
+			sourcePkOrUniqueFieldMap=new HashMap<>();
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"sourcePkOrUniqueFieldMap",sourcePkOrUniqueFieldMap);
+		}
+		@DisplayName("test getPkOrUniqueValueKey value not exist")
+		@Test
+		void test1(){
+			List<String> pkList=new ArrayList<>();
+			pkList.add("id");
+			sourcePkOrUniqueFieldMap.put("sourceId",pkList);
+			doCallRealMethod().when(mockHazelcastMergeNode).getPkOrUniqueValueKey(anyMap(),anyString(),any(ConstructIMap.class));
+			Map<String,Object> after=new HashMap<>();
+			after.put("cid",1);
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getPkOrUniqueValueKey(after, "sourceId", constructIMap);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.PK_OR_UNIQUE_VALUE_NOT_EXISTS);
+		}
+	}
+	@Nested
+	class getAndUpdateJoinKeyCacheClass{
+		HazelcastMergeNode mockHazelcastMergeNode;
+		ConstructIMap<Document> constructIMap;
+		Map<String, ConstructIMap<Document>> checkJoinKeyUpdateCacheMap;
+		@BeforeEach
+		void init(){
+			checkJoinKeyUpdateCacheMap=new HashMap<>();
+			mockHazelcastMergeNode=mock(HazelcastMergeNode.class);
+			constructIMap=mock(ConstructIMap.class);
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"checkJoinKeyUpdateCacheMap",checkJoinKeyUpdateCacheMap);
+			Node tableNode=mock(TableNode.class);
+			when(tableNode.getName()).thenReturn("nodeName");
+			when(mockHazelcastMergeNode.getPreNode(anyString())).thenReturn(tableNode);
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("Mongo ExteralStorageDto");
+			ReflectionTestUtils.setField(mockHazelcastMergeNode,"externalStorageDto",externalStorageDto);
+		}
+		@DisplayName("test throw GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_GET_IMAP ")
+		@Test
+		void test1(){
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			tapdataEvent.addNodeId("nodeId");
+			doCallRealMethod().when(mockHazelcastMergeNode).getPreNodeId(any());
+			doCallRealMethod().when(mockHazelcastMergeNode).getAndUpdateJoinKeyCache(any());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getAndUpdateJoinKeyCache(tapdataEvent);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_CANNOT_GET_IMAP);
+		}
+		@DisplayName("test throw GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_FIND_BY_PK_FAILED exception")
+		@Test
+		void test2() throws Exception {
+			checkJoinKeyUpdateCacheMap.put("nodeId", constructIMap);
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			tapdataEvent.addNodeId("nodeId");
+			Map<String,Object> after=new HashMap<>();
+			after.put("id",1);
+			TapInsertRecordEvent tapInsertRecordEvent = TapInsertRecordEvent.create();
+			tapInsertRecordEvent.setAfter(after);
+			tapdataEvent.setTapEvent(tapInsertRecordEvent);
+			when(mockHazelcastMergeNode.getPkOrUniqueValueKey(anyMap(),anyString(),any(ConstructIMap.class))).thenReturn("1");
+			when(constructIMap.find(anyString())).thenThrow(new RuntimeException("throw failed"));
+			doCallRealMethod().when(mockHazelcastMergeNode).getPreNodeId(any());
+			doCallRealMethod().when(mockHazelcastMergeNode).getAndUpdateJoinKeyCache(any());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getAndUpdateJoinKeyCache(tapdataEvent);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_FIND_BY_PK_FAILED);
+		}
+		@DisplayName("test getAndUpdateJoinKeyCache throw GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_UPSERT_FAILED")
+		@Test
+		void test3() throws Exception {
+			checkJoinKeyUpdateCacheMap.put("nodeId", constructIMap);
+			TapdataEvent tapdataEvent=new TapdataEvent();
+			tapdataEvent.addNodeId("nodeId");
+			Map<String,Object> after=new HashMap<>();
+			after.put("id",1);
+			TapInsertRecordEvent tapInsertRecordEvent = TapInsertRecordEvent.create();
+			tapInsertRecordEvent.setAfter(after);
+			tapdataEvent.setTapEvent(tapInsertRecordEvent);
+			when(mockHazelcastMergeNode.getPkOrUniqueValueKey(anyMap(),anyString(),any(ConstructIMap.class))).thenReturn("1");
+			Document document = new Document();
+			when(constructIMap.find(anyString())).thenReturn(document);
+			when(constructIMap.upsert(any(),any())).thenThrow(new RuntimeException("upsertFailed"));
+			doCallRealMethod().when(mockHazelcastMergeNode).getPreNodeId(any());
+			doCallRealMethod().when(mockHazelcastMergeNode).getAndUpdateJoinKeyCache(any());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				mockHazelcastMergeNode.getAndUpdateJoinKeyCache(tapdataEvent);
+			});
+			assertEquals(tapCodeException.getCode(),TaskMergeProcessorExCode_16.GET_AND_UPDATE_JOIN_KEY_CACHE_FAILED_UPSERT_FAILED);
+		}
+
+	}
 
 	@Nested
 	@DisplayName("Method recursiveLookup test")
@@ -2219,6 +2631,46 @@ void testUpdateWriteLookupOnlyOneRow() {
 
 			verify(nodeLogger, times(0)).warn(eq("Update write merge lookup, find more than one row, lookup table: {}, join key value: {}, will use first row: {}"), any(Object[].class));
 		}
+		@Test
+		@SneakyThrows
+		@DisplayName("test update write lookup and return only one row")
+		void testUpdateWriteLookupOnlyOneRowThrowError() {
+			MergeTableProperties mergeTableProperties = new MergeTableProperties();
+			MergeTableProperties mergeTableProperties1 = new MergeTableProperties();
+			List<MergeTableProperties> mergeTablePropertiesList = new ArrayList<>();
+			mergeTablePropertiesList.add(mergeTableProperties);
+			mergeTableProperties1.setChildren(mergeTablePropertiesList);
+			mergeTableProperties1.setId("2");
+			mergeTableProperties.setId("1");
+			mergeTableProperties.setMergeType(MergeTableProperties.MergeType.updateWrite);
+			mergeTableProperties.setTargetPath("test");
+			Map<String, String> joinKeyMap = new HashMap<>();
+			joinKeyMap.put("source", "id");
+			joinKeyMap.put("target", "id");
+			List<Map<String, String>> joinKeys = new ArrayList<>();
+			joinKeys.add(joinKeyMap);
+			mergeTableProperties.setJoinKeys(joinKeys);
+			Map<String, Object> data = new HashMap<>();
+			data.put("id", 1);
+			ConstructIMap<Document> constructIMap = mock(ConstructIMap.class);
+			when(constructIMap.getName()).thenReturn("test");
+			when(constructIMap.find(anyString())).thenThrow(new RuntimeException("connect failed"));
+			Node preNode = mock(TableNode.class);
+			when(((TableNode) preNode).getTableName()).thenReturn("test");
+			Node fatherPreNode = mock(TableNode.class);
+			when(((TableNode) preNode).getTableName()).thenReturn("fathertest");
+			ExternalStorageDto externalStorageDto=new ExternalStorageDto();
+			externalStorageDto.setType("Mongo");
+			externalStorageDto.setName("externalName");
+			ReflectionTestUtils.setField(hazelcastMergeNode,"externalStorageDto",externalStorageDto);
+			doReturn(preNode).when(hazelcastMergeNode).getPreNode("1");
+			doReturn(fatherPreNode).when(hazelcastMergeNode).getPreNode("2");
+			doReturn(constructIMap).when(hazelcastMergeNode).getHazelcastConstruct(anyString());
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.recursiveLookup(mergeTableProperties1, data, true);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.LOOK_UP_FIND_BY_JOIN_KEY_FAILED,tapCodeException.getCode());
+		}
 
 		@Test
 		@SneakyThrows
@@ -2328,6 +2780,61 @@ void testNodeIsNotDisabled(){
 			hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
 			Assertions.assertEquals(1,sourcePkOrUniqueFieldMap.size());
 		}
+		@Test
+		void testNoPrimaryKeyThrowException(){
+			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(null);
+			TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+				hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
+			});
+			assertEquals(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_PRIMARY_KEY,tapCodeException.getCode());
+		}
+
+		@Test
+		void test1(){
+			MergeTableProperties mergeTableProperties = new MergeTableProperties();
+			mergeTableProperties.setId("2cbc1a4d-906d-4b32-9cf4-6596ed4bd0e4");
+			mergeTableProperties.setMergeType(MergeTableProperties.MergeType.updateIntoArray);
+			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"));
+			try{
+				hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
+			}catch (TapCodeException e){
+				assertEquals(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NO_ARRAY_KEY,e.getCode());
+			}
+
+		}
 
 	}
 
@@ -2469,4 +2976,37 @@ void testCloud(){
 			}
 		}
 	}
+	@DisplayName("测试错误码，表名不能为空")
+	@Test
+	void test2(){
+		try{
+			TableNode tableNode=new TableNode();
+			hazelcastMergeNode.getTableName(tableNode);
+		}catch (TapCodeException e){
+			assertEquals(TaskMergeProcessorExCode_16.TABLE_NAME_CANNOT_BE_BLANK,e.getCode());
+		}
+	}
+	@DisplayName("测试错误码，连接ID不能为空")
+	@Test
+	void test3(){
+		try{
+			TableNode tableNode=new TableNode();
+			hazelcastMergeNode.getConnectionId(tableNode);
+		}catch (TapCodeException e){
+			assertEquals(TaskMergeProcessorExCode_16.CONNECTION_ID_CANNOT_BE_BLANK,e.getCode());
+		}
+	}
+	@Test
+	void test4(){
+		MergeTableProperties mergeTableProperties=new MergeTableProperties();
+		mergeTableProperties.setId("sourceTableId");
+		List<MergeTableProperties> mergeTablePropertiesList =new ArrayList<>();
+		mergeTablePropertiesList.add(mergeTableProperties);
+		try{
+			hazelcastMergeNode.initSourcePkOrUniqueFieldMap(mergeTablePropertiesList);
+		}catch (TapCodeException e){
+			assertEquals(TaskMergeProcessorExCode_16.TAP_MERGE_TABLE_NODE_NOT_FOUND,e.getCode());
+		}
+
+	}
 }
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/constant/CommonUtilTest.java b/iengine/iengine-common/src/test/java/com/tapdata/constant/CommonUtilTest.java
new file mode 100644
index 0000000000..ed4c565e21
--- /dev/null
+++ b/iengine/iengine-common/src/test/java/com/tapdata/constant/CommonUtilTest.java
@@ -0,0 +1,170 @@
+package com.tapdata.constant;
+
+import com.tapdata.entity.values.BooleanNotExist;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class CommonUtilTest {
+    @Nested
+    class compareBooleanTest{
+        @DisplayName("test one value is boolean")
+        @Test
+        void test1(){
+            boolean result = CommonUtil.compareBoolean(true, null);
+            assertEquals(false,result);
+        }
+        @DisplayName("test one value is boolean")
+        @Test
+        void test5(){
+            boolean result = CommonUtil.compareBoolean(null, true);
+            assertEquals(false,result);
+        }
+        @DisplayName("test two value is null")
+        @Test
+        void test2(){
+            boolean result = CommonUtil.compareBoolean(null, null);
+            assertEquals(true,result);
+        }
+        @DisplayName("test two value is boolean")
+        @Test
+        void test3(){
+            boolean result = CommonUtil.compareBoolean(true, false);
+            assertEquals(false,result);
+        }
+        @DisplayName("test one value is 1")
+        @Test
+        void test4(){
+            boolean result = CommonUtil.compareBoolean(true, "1");
+            assertEquals(true,result);
+        }
+    }
+    @Nested
+    class testToBoolean{
+        @DisplayName("test Number not 1 or 0")
+        @Test
+        void test1(){
+            Object aBoolean = CommonUtil.toBoolean(2);
+            assertEquals(true, aBoolean instanceof BooleanNotExist);
+        }
+        @DisplayName("test Number 1")
+        @Test
+        void test2(){
+            Object aBoolean = CommonUtil.toBoolean(1);
+            assertEquals(true,aBoolean);
+        }
+        @DisplayName("test Number 0")
+        @Test
+        void test3(){
+            Object aBoolean = CommonUtil.toBoolean(0);
+            assertEquals(false,aBoolean);
+        }
+        @DisplayName("test Number 0.0")
+        @Test
+        void test4(){
+            Object aBoolean = CommonUtil.toBoolean(0.0);
+            assertEquals(false,aBoolean);
+        }
+        @DisplayName("test BigDecimal 1")
+        @Test
+        void test5(){
+            Object aBoolean = CommonUtil.toBoolean(new BigDecimal("1"));
+            assertEquals(true,aBoolean);
+        }
+        @DisplayName("test BigDecimal 0")
+        @Test
+        void test6(){
+            Object aBoolean = CommonUtil.toBoolean(new BigDecimal("0"));
+            assertEquals(false,aBoolean);
+        }
+        @DisplayName("test BigDecimal not 0 or 1")
+        @Test
+        void test7(){
+            Object aBoolean = CommonUtil.toBoolean(new BigDecimal("2"));
+            assertEquals(true,aBoolean instanceof BooleanNotExist);
+        }
+        @DisplayName("test string true")
+        @Test
+        void test9(){
+            Object aBoolean = CommonUtil.toBoolean("true");
+            assertEquals(true,aBoolean);
+        }
+        @DisplayName("test string True")
+        @Test
+        void test10(){
+            Object aBoolean = CommonUtil.toBoolean("True");
+            assertEquals(true,aBoolean);
+        }
+        @DisplayName("test string False")
+        @Test
+        void test11(){
+            Object aBoolean = CommonUtil.toBoolean("False");
+            assertEquals(false,aBoolean);
+        }
+        @DisplayName("test string 0")
+        @Test
+        void test12(){
+            Object aBoolean = CommonUtil.toBoolean("0");
+            assertEquals(false,aBoolean);
+        }
+        @DisplayName("test string 0")
+        @Test
+        void test13(){
+            Object aBoolean = CommonUtil.toBoolean("1");
+            assertEquals(true,aBoolean);
+        }
+        @DisplayName("test string not 0,false,1,true")
+        @Test
+        void test14(){
+            Object aBoolean = CommonUtil.toBoolean("2");
+            assertEquals(true,aBoolean instanceof BooleanNotExist);
+        }
+    }
+    @Nested
+    class compareObjectsTest{
+        @DisplayName("test val1 is boolean and val2 can cast to boolean")
+        @Test
+        void test1() {
+            Object[] val1 = new Object[10];
+            Object[] val2 = new Object[10];
+            val1[0] = true;
+            val2[0] = "1";
+            int result = CommonUtil.compareObjects(val1, val2);
+            assertEquals(0, result);
+        }
+        @DisplayName("test val1 is boolean and val2 can not cast to boolean")
+        @Test
+        void test2(){
+            Object[] val1 = new Object[10];
+            Object[] val2 = new Object[10];
+            val1[0] = true;
+            val2[0] = "2";
+            int result = CommonUtil.compareObjects(val1, val2);
+            assertEquals(1, result);
+        }
+        @DisplayName("test val2 is boolean and val1 can not cast to boolean")
+        @Test
+        void test3(){
+            Object[] val1 = new Object[10];
+            Object[] val2 = new Object[10];
+            val1[0] = 2;
+            val2[0] = true;
+            int result = CommonUtil.compareObjects(val1, val2);
+            assertEquals(-1, result);
+        }
+        @DisplayName("test val2 is boolean and val1 can not cast to boolean")
+        @Test
+        void test4(){
+            Object[] val1 = new Object[10];
+            Object[] val2 = new Object[10];
+            val1[0] = "1";
+            val2[0] = true;
+            int result = CommonUtil.compareObjects(val1, val2);
+            assertEquals(0, result);
+        }
+    }
+}
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/processor/ScriptUtilTest.java b/iengine/iengine-common/src/test/java/com/tapdata/processor/ScriptUtilTest.java
index d1af8ea891..285f315e35 100644
--- a/iengine/iengine-common/src/test/java/com/tapdata/processor/ScriptUtilTest.java
+++ b/iengine/iengine-common/src/test/java/com/tapdata/processor/ScriptUtilTest.java
@@ -1,5 +1,6 @@
 package com.tapdata.processor;
 
+import com.oracle.truffle.js.scriptengine.GraalJSScriptEngine;
 import com.tapdata.cache.ICacheGetter;
 import com.tapdata.entity.Connections;
 import com.tapdata.entity.Job;
@@ -8,22 +9,25 @@
 import com.tapdata.processor.constant.JSEngineEnum;
 import com.tapdata.processor.error.ScriptProcessorExCode_30;
 import io.tapdata.entity.logger.Log;
+import io.tapdata.entity.script.ScriptFactory;
+import io.tapdata.entity.script.ScriptOptions;
+import io.tapdata.entity.utils.InstanceFactory;
 import io.tapdata.exception.TapCodeException;
 import lombok.SneakyThrows;
 import org.apache.logging.log4j.core.Logger;
 import org.junit.Assert;
 import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
 import org.mockito.internal.verification.Times;
 
 import javax.script.Invocable;
-import java.io.IOException;
+import javax.script.ScriptEngine;
+import javax.script.ScriptException;
 import java.net.URL;
-import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Consumer;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -70,10 +74,11 @@ public void testUrlClassLoaderException(){
         assertEquals(ScriptProcessorExCode_30.GET_SCRIPT_ENGINE_ERROR,tapCodeException.getCode());
 
     }
+
     @Nested
     public class InvokeScriptWithTagTest{
         private Map<String,Object> input;
-        private Invocable engine;
+        private Invocable engine = mock(Invocable.class);
         private MessageEntity message;
         @BeforeEach
         @SneakyThrows
@@ -127,6 +132,84 @@ public void testInvokeScriptWithNull(){
             verify(message,new Times(2)).getBefore();
             verify(message,new Times(1)).getAfter();
         }
+        @SneakyThrows
+        @Test
+        public void testInvokeScriptWithoutEngine(){
+            TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+                ScriptUtil.invokeScript(null, "process", message
+                        , mock(Connections.class), mock(Connections.class), mock(Job.class), input, mock(Logger.class), null);
+            });
+            assertEquals(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED_ENGINE_NULL,tapCodeException.getCode());
+        }
+        @SneakyThrows
+        @DisplayName("test Invoke Script function appear INVOKE_SCRIPT_FAILED")
+        @Test
+        public void test1(){
+            when(message.getAfter()).thenReturn(mock(HashMap.class));
+            Invocable engine= mock(GraalJSScriptEngine.class);
+            when(engine.invokeFunction(anyString(),any())).thenThrow(new ScriptException("script failed"));
+            TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+                ScriptUtil.invokeScript(engine, "process", message
+                        , mock(Connections.class), mock(Connections.class), mock(Job.class), input, mock(Logger.class), null);
+            });
+            assertEquals(ScriptProcessorExCode_30.INVOKE_SCRIPT_FAILED,tapCodeException.getCode());
+
+        }
+        @Nested
+        public class GetPyEngineTest{
 
+            @SneakyThrows
+            @Test
+            @DisplayName("test get python script with GET_PYTHON_ENGINE_FAILED exception")
+            void test1() {
+                String initPythonBuildInMethod = ScriptUtil.initPythonBuildInMethod(null, null, ScriptUtilTest.class.getClassLoader(), (classLoader) -> {
+                });
+                try (MockedStatic<InstanceFactory> instanceFactoryMockedStatic = mockStatic(InstanceFactory.class)) {
+                    ScriptFactory scriptFactory = mock(ScriptFactory.class);
+                    ScriptEngine scriptEngine = mock(ScriptEngine.class);
+                    when(scriptEngine.eval(initPythonBuildInMethod)).thenThrow(new ScriptException("eval error"));
+                    when(scriptFactory.create(anyString(),any(ScriptOptions.class))).thenReturn(scriptEngine);
+                    instanceFactoryMockedStatic.when(()->{InstanceFactory.instance(any(Class.class),anyString());}).thenReturn(scriptFactory);
+                    TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+                        Invocable pyEngine = ScriptUtil.getPyEngine(
+                                ScriptFactory.TYPE_PYTHON,
+                                null,
+                                null, //javaScriptFunctions,
+                                null,
+                                null,
+                                null,
+                                null,
+                                mock(Log.class),
+                                ScriptUtilTest.class.getClassLoader());
+                    });
+                    assertEquals(ScriptProcessorExCode_30.GET_PYTHON_ENGINE_FAILED,tapCodeException.getCode());
+                }
+            }
+            @SneakyThrows
+            @Test
+            void test2(){
+                try (MockedStatic<InstanceFactory> instanceFactoryMockedStatic = mockStatic(InstanceFactory.class)) {
+                    ScriptFactory scriptFactory = mock(ScriptFactory.class);
+                    ScriptEngine scriptEngine = mock(ScriptEngine.class);
+                    String scripts = "" + System.lineSeparator() + ScriptUtil.handlePyScript(ScriptUtil.DEFAULT_PY_SCRIPT);
+                    when(scriptEngine.eval(scripts)).thenThrow(new ScriptException("eval error"));
+                    when(scriptFactory.create(anyString(),any(ScriptOptions.class))).thenReturn(scriptEngine);
+                    instanceFactoryMockedStatic.when(()->{InstanceFactory.instance(any(Class.class),anyString());}).thenReturn(scriptFactory);
+                    TapCodeException tapCodeException = assertThrows(TapCodeException.class, () -> {
+                        Invocable pyEngine = ScriptUtil.getPyEngine(
+                                ScriptFactory.TYPE_PYTHON,
+                                null,
+                                null, //javaScriptFunctions,
+                                null,
+                                null,
+                                null,
+                                null,
+                                mock(Log.class),
+                                ScriptUtilTest.class.getClassLoader());
+                    });
+                    assertEquals(ScriptProcessorExCode_30.GET_PYTHON_ENGINE_FAILED,tapCodeException.getCode());
+                }
+            }
+        }
     }
 }
diff --git a/iengine/modules/milestone-module/src/test/java/io/tapdata/milestone/MilestoneAspectTaskTest.java b/iengine/modules/milestone-module/src/test/java/io/tapdata/milestone/MilestoneAspectTaskTest.java
index cc521465cf..b2a4f8aacc 100644
--- a/iengine/modules/milestone-module/src/test/java/io/tapdata/milestone/MilestoneAspectTaskTest.java
+++ b/iengine/modules/milestone-module/src/test/java/io/tapdata/milestone/MilestoneAspectTaskTest.java
@@ -16,6 +16,7 @@
 import io.tapdata.aspect.taskmilestones.*;
 import io.tapdata.entity.aspect.Aspect;
 import io.tapdata.entity.aspect.AspectInterceptResult;
+import io.tapdata.entity.error.CoreException;
 import io.tapdata.entity.logger.Log;
 import io.tapdata.entity.simplify.pretty.TypeHandlers;
 import io.tapdata.exception.TapCodeException;
@@ -1394,6 +1395,14 @@ void testSetError() {
 		aspect.error(new TapCodeException("test1"));
 		milestoneAspectTask.setError(aspect, entity);
 		Assertions.assertEquals("test", entity.getErrorCode());
+
+		entity = new MilestoneEntity();
+		aspect = new WriteErrorAspect();
+		CoreException coreError = new CoreException("test");
+		coreError.setCode(1);
+		aspect.error(coreError);
+		milestoneAspectTask.setError(aspect, entity);
+		Assertions.assertEquals("1", entity.getErrorCode());
 	}
 
 	@Test
@@ -1411,9 +1420,22 @@ void testSetError_1() {
 		Assertions.assertEquals("test", entity.getErrorCode());
 
 		aspect = new EngineDeductionAspect();
-		aspect.error(new TapCodeException("test1"));
+		TapCodeException error = new TapCodeException("test1");
+		error.dynamicDescriptionParameters("test1", "test2");
+		aspect.error(error);
 		milestoneAspectTask.setError(aspect, entity);
 		Assertions.assertEquals("test", entity.getErrorCode());
+		Assertions.assertNotNull(entity.getDynamicDescriptionParameters());
+		Assertions.assertEquals(2, entity.getDynamicDescriptionParameters().length);
+
+		entity = new MilestoneEntity();
+		aspect = new EngineDeductionAspect();
+		CoreException coreError = new CoreException("test1");
+		coreError.setCode(1);
+		aspect.error(coreError);
+		milestoneAspectTask.setError(aspect, entity);
+		Assertions.assertEquals("1", entity.getErrorCode());
+		Assertions.assertNull(entity.getDynamicDescriptionParameters());
 	}
 
 }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/TaskLoggerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/TaskLoggerTest.java
deleted file mode 100644
index 20c0dc2c71..0000000000
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/TaskLoggerTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-package io.tapdata.observable;
-
-import io.tapdata.observable.logging.TaskLogger;
-import io.tapdata.observable.logging.appender.AppenderFactory;
-import io.tapdata.observable.logging.appender.FileAppender;
-import io.tapdata.observable.logging.appender.ObsHttpTMAppender;
-import io.tapdata.observable.logging.util.Conf.LogConfiguration;
-import org.apache.logging.log4j.core.appender.RollingFileAppender;
-import org.apache.logging.log4j.core.appender.rolling.CompositeTriggeringPolicy;
-import org.apache.logging.log4j.core.appender.rolling.TimeBasedTriggeringPolicy;
-import org.apache.logging.log4j.core.appender.rolling.TriggeringPolicy;
-import org.junit.jupiter.api.DisplayName;
-import org.junit.jupiter.api.Test;
-import org.mockito.MockedStatic;
-import org.springframework.test.util.ReflectionTestUtils;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.*;
-
-public class TaskLoggerTest {
-    @DisplayName("test Refresh FileAppender")
-    @Test
-    void test1(){
-        try(MockedStatic<AppenderFactory> appenderFactoryMockedStatic = mockStatic(AppenderFactory.class)){
-            List<io.tapdata.observable.logging.appender.Appender<?>> tapObsAppenders = new ArrayList<>();
-            LogConfiguration logConfiguration = LogConfiguration.builder().logSaveTime(180).logSaveSize(10).logSaveCount(100).build();
-            AppenderFactory appenderFactory = mock(AppenderFactory.class);
-            when(AppenderFactory.getInstance()).thenReturn(appenderFactory);
-            TaskLogger taskLogger = mock(TaskLogger.class);
-            TimeBasedTriggeringPolicy timeBasedTriggeringPolicy = TimeBasedTriggeringPolicy.newBuilder().withInterval(1).withModulate(true).build();
-            RollingFileAppender rollingFileAppender = RollingFileAppender.newBuilder()
-                    .setName("rollingFileAppender")
-                    .withFileName("./tapdata-agent.log")
-                    .withFilePattern("./tapdata-agent-%i.log.%d{yyyyMMdd}.gz")
-                    .withPolicy(timeBasedTriggeringPolicy)
-                    .build();
-            FileAppender fileAppender = mock(FileAppender.class);
-            ReflectionTestUtils.setField(fileAppender,"rollingFileAppender",rollingFileAppender);
-            doCallRealMethod().when(fileAppender).getRollingFileAppender();
-            tapObsAppenders.add(fileAppender);
-            ReflectionTestUtils.setField(taskLogger,"tapObsAppenders",tapObsAppenders);
-            ReflectionTestUtils.setField(taskLogger,"logAppendFactory",appenderFactory);
-            doCallRealMethod().when(taskLogger).refreshFileAppender(logConfiguration);
-            taskLogger.refreshFileAppender(logConfiguration);
-            assertEquals("./tapdata-agent.log",rollingFileAppender.getFileName());
-            assertEquals("rollingFileAppender",rollingFileAppender.getName());
-            TriggeringPolicy triggeringPolicy = rollingFileAppender.getTriggeringPolicy();
-            boolean updateSuccess= triggeringPolicy instanceof CompositeTriggeringPolicy;
-            assertEquals(true,updateSuccess);
-        }
-
-    }
-
-    @DisplayName("test appenders list no have FileAppender ")
-    @Test
-    void test2() {
-        LogConfiguration logConfiguration = LogConfiguration.builder().logSaveTime(180).logSaveSize(10).logSaveCount(100).build();
-        TaskLogger taskLogger = mock(TaskLogger.class);
-        List<io.tapdata.observable.logging.appender.Appender<?>> tapObsAppenders = new ArrayList<>();
-        ObsHttpTMAppender obsHttpTMAppender = mock(ObsHttpTMAppender.class);
-        tapObsAppenders.add(obsHttpTMAppender);
-        ReflectionTestUtils.setField(taskLogger, "tapObsAppenders", tapObsAppenders);
-        doCallRealMethod().when(taskLogger).refreshFileAppender(logConfiguration);
-        assertEquals(1, tapObsAppenders.size());
-        assertEquals(true, obsHttpTMAppender == tapObsAppenders.get(0));
-    }
-}
\ No newline at end of file
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/BlankObsLoggerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/BlankObsLoggerTest.java
new file mode 100644
index 0000000000..11ef758232
--- /dev/null
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/BlankObsLoggerTest.java
@@ -0,0 +1,20 @@
+package io.tapdata.observable.logging;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/11/14 18:52
+ */
+public class BlankObsLoggerTest {
+
+    @Test
+    void testTrace() {
+        BlankObsLogger blankObsLogger = new BlankObsLogger();
+        Assertions.assertDoesNotThrow(() -> {
+            blankObsLogger.trace(null, null);
+        });
+    }
+
+}
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/LoggingAspectTaskTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/LoggingAspectTaskTest.java
new file mode 100644
index 0000000000..c3e280a73b
--- /dev/null
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/LoggingAspectTaskTest.java
@@ -0,0 +1,73 @@
+package io.tapdata.observable.logging;
+
+import com.tapdata.constant.BeanUtil;
+import com.tapdata.entity.task.context.DataProcessorContext;
+import com.tapdata.mongo.ClientMongoOperator;
+import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
+import io.tapdata.aspect.CreateTableFuncAspect;
+import io.tapdata.common.SettingService;
+import io.tapdata.entity.event.ddl.table.TapCreateTableEvent;
+import io.tapdata.entity.schema.TapTable;
+import io.tapdata.pdk.apis.functions.connector.target.CreateTableOptions;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.mockito.Mockito.*;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/11/15 10:24
+ */
+public class LoggingAspectTaskTest {
+
+    @Test
+    void testHandlerCreateTableFun() {
+
+        LoggingAspectTask aspectTask = new LoggingAspectTask();
+
+        CreateTableFuncAspect aspect = new CreateTableFuncAspect();
+
+        DataProcessorContext dataProcessContext = mock(DataProcessorContext.class);
+        Node node = new DatabaseNode();
+        node.setId("nodeId");
+        node.setName("nodeName");
+        when(dataProcessContext.getNode()).thenReturn(node);
+        aspect.setDataProcessorContext(dataProcessContext);
+        aspect.setCreateTableEvent(new TapCreateTableEvent());
+        aspect.getCreateTableEvent().setTable(new TapTable());
+
+        ConcurrentHashMap<String, ObsLogger> map = mock(ConcurrentHashMap.class);
+        ObsLogger logger = mock(ObsLogger.class);
+        when(map.computeIfAbsent(anyString(), any())).thenReturn(logger);
+        ReflectionTestUtils.setField(aspectTask, "nodeLoggerMap", map);
+
+        try (MockedStatic<BeanUtil> beanUtilMock = mockStatic(BeanUtil.class)) {
+
+            beanUtilMock.when(() -> BeanUtil.getBean(any())).thenAnswer(answer -> {
+                Class cls = answer.getArgument(1);
+                return mock(cls);
+            });
+
+            LoggingAspectTask spyAspectTask = spy(aspectTask);
+
+            aspect.state(CreateTableFuncAspect.STATE_END);
+            spyAspectTask.handleCreateTableFuc(aspect);
+            aspect.setCreateTableOptions(CreateTableOptions.create());
+            spyAspectTask.handleCreateTableFuc(aspect);
+            aspect.getCreateTableOptions().tableExists(true);
+            spyAspectTask.handleCreateTableFuc(aspect);
+            aspect.getCreateTableOptions().tableExists(false);
+            spyAspectTask.handleCreateTableFuc(aspect);
+
+            verify(logger, times(1)).info(anyString(), any());
+
+        }
+
+    }
+
+}
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerNodeProxyTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerNodeProxyTest.java
index 9ec75af8a6..ae3fd7b373 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerNodeProxyTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerNodeProxyTest.java
@@ -1,15 +1,13 @@
 package io.tapdata.observable.logging;
 
 import com.tapdata.tm.commons.schema.MonitoringLogsDto;
-import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.springframework.util.CollectionUtils;
 
 import java.util.Collections;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 /**
  * @author lg&lt;lirufei0808@gmail.com&gt;
@@ -47,4 +45,17 @@ public void testBaseBuilder() {
         Assertions.assertEquals("taskId", dto.getTaskId());
         Assertions.assertTrue(CollectionUtils.isEmpty(dto.getLogTags()));
     }
+
+    @Test
+    void testTrace() {
+        TaskLogger taskLogger = mock(TaskLogger.class);
+        TaskLoggerNodeProxy logProxy = new TaskLoggerNodeProxy()
+                .withTaskLogger(taskLogger)
+                .withNode("NodeId", "NodeName")
+                .withTags(Collections.singletonList("tag1"));
+
+        logProxy.trace(MonitoringLogsDto::builder, "test {}", 1);
+
+        verify(taskLogger, times(1)).trace(any(), anyString(), any());
+    }
 }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerTest.java
new file mode 100644
index 0000000000..4fc510f993
--- /dev/null
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/TaskLoggerTest.java
@@ -0,0 +1,199 @@
+package io.tapdata.observable.logging;
+
+import com.tapdata.constant.ConfigurationCenter;
+import com.tapdata.tm.commons.schema.MonitoringLogsDto;
+import com.tapdata.tm.commons.task.dto.TaskDto;
+import io.tapdata.flow.engine.V2.entity.GlobalConstant;
+import io.tapdata.observable.logging.appender.AppenderFactory;
+import io.tapdata.observable.logging.appender.FileAppender;
+import io.tapdata.observable.logging.appender.ObsHttpTMAppender;
+import io.tapdata.observable.logging.util.Conf.LogConfiguration;
+import org.apache.logging.log4j.core.appender.RollingFileAppender;
+import org.apache.logging.log4j.core.appender.rolling.CompositeTriggeringPolicy;
+import org.apache.logging.log4j.core.appender.rolling.TimeBasedTriggeringPolicy;
+import org.apache.logging.log4j.core.appender.rolling.TriggeringPolicy;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
+public class TaskLoggerTest {
+    @DisplayName("test Refresh FileAppender")
+    @Test
+    void test1(){
+        try(MockedStatic<AppenderFactory> appenderFactoryMockedStatic = mockStatic(AppenderFactory.class)){
+            List<io.tapdata.observable.logging.appender.Appender<?>> tapObsAppenders = new ArrayList<>();
+            LogConfiguration logConfiguration = LogConfiguration.builder().logSaveTime(180).logSaveSize(10).logSaveCount(100).build();
+            AppenderFactory appenderFactory = mock(AppenderFactory.class);
+            when(AppenderFactory.getInstance()).thenReturn(appenderFactory);
+            TaskLogger taskLogger = mock(TaskLogger.class);
+            TimeBasedTriggeringPolicy timeBasedTriggeringPolicy = TimeBasedTriggeringPolicy.newBuilder().withInterval(1).withModulate(true).build();
+            RollingFileAppender rollingFileAppender = RollingFileAppender.newBuilder()
+                    .setName("rollingFileAppender")
+                    .withFileName("./tapdata-agent.log")
+                    .withFilePattern("./tapdata-agent-%i.log.%d{yyyyMMdd}.gz")
+                    .withPolicy(timeBasedTriggeringPolicy)
+                    .build();
+            FileAppender fileAppender = mock(FileAppender.class);
+            ReflectionTestUtils.setField(fileAppender,"rollingFileAppender",rollingFileAppender);
+            doCallRealMethod().when(fileAppender).getRollingFileAppender();
+            tapObsAppenders.add(fileAppender);
+            ReflectionTestUtils.setField(taskLogger,"tapObsAppenders",tapObsAppenders);
+            ReflectionTestUtils.setField(taskLogger,"logAppendFactory",appenderFactory);
+            doCallRealMethod().when(taskLogger).refreshFileAppender(logConfiguration);
+            taskLogger.refreshFileAppender(logConfiguration);
+            assertEquals("./tapdata-agent.log",rollingFileAppender.getFileName());
+            assertEquals("rollingFileAppender",rollingFileAppender.getName());
+            TriggeringPolicy triggeringPolicy = rollingFileAppender.getTriggeringPolicy();
+            boolean updateSuccess= triggeringPolicy instanceof CompositeTriggeringPolicy;
+            assertEquals(true,updateSuccess);
+        }
+
+    }
+
+    @DisplayName("test appenders list no have FileAppender ")
+    @Test
+    void test2() {
+        LogConfiguration logConfiguration = LogConfiguration.builder().logSaveTime(180).logSaveSize(10).logSaveCount(100).build();
+        TaskLogger taskLogger = mock(TaskLogger.class);
+        List<io.tapdata.observable.logging.appender.Appender<?>> tapObsAppenders = new ArrayList<>();
+        ObsHttpTMAppender obsHttpTMAppender = mock(ObsHttpTMAppender.class);
+        tapObsAppenders.add(obsHttpTMAppender);
+        ReflectionTestUtils.setField(taskLogger, "tapObsAppenders", tapObsAppenders);
+        doCallRealMethod().when(taskLogger).refreshFileAppender(logConfiguration);
+        assertEquals(1, tapObsAppenders.size());
+        assertEquals(true, obsHttpTMAppender == tapObsAppenders.get(0));
+    }
+
+    @Nested
+    class testTaskLogger {
+        @Test
+        void testEnableDebug() {
+            TaskLogger taskLogger = mock(TaskLogger.class);
+            doCallRealMethod().when(taskLogger).withTaskLogSetting(anyString(), anyLong(), anyLong());
+            doCallRealMethod().when(taskLogger).isDebugEnabled();
+            doCallRealMethod().when(taskLogger).noNeedLog(any());
+
+            taskLogger.withTaskLogSetting("INFO", 1L, 1L);
+            Assertions.assertFalse((Boolean) ReflectionTestUtils.getField(taskLogger, "enableDebugLogger"));
+
+            taskLogger.withTaskLogSetting("DEBUG", 1000L, 1000L);
+            Assertions.assertTrue((Boolean)ReflectionTestUtils.getField(taskLogger, "enableDebugLogger"));
+        }
+
+        @Test
+        void testNoNeedLog() {
+            TaskLogger taskLogger = mock(TaskLogger.class);
+            doCallRealMethod().when(taskLogger).withTaskLogSetting(anyString(), anyLong(), anyLong());
+            doCallRealMethod().when(taskLogger).noNeedLog(any());
+
+            taskLogger.withTaskLogSetting("DEBUG", 10L, 60L);
+
+            for (int i = 0; i < 100; i++) {
+                Assertions.assertFalse(taskLogger.noNeedLog("TRACE"));
+            }
+            for (int i = 0; i < 10; i++) {
+                Assertions.assertFalse(taskLogger.noNeedLog("DEBUG"));
+            }
+            Assertions.assertTrue(taskLogger.noNeedLog("DEBUG"));
+        }
+
+        @Test
+        void testTrace() {
+            TaskLogger taskLogger = mock(TaskLogger.class);
+            doCallRealMethod().when(taskLogger).withTaskLogSetting(anyString(), anyLong(), anyLong());
+            doCallRealMethod().when(taskLogger).noNeedLog(any());
+            doCallRealMethod().when(taskLogger).trace(any(), anyString(), any());
+
+            AppenderFactory logAppendFactory = mock(AppenderFactory.class);
+            ReflectionTestUtils.setField(taskLogger, "logAppendFactory", logAppendFactory);
+
+            taskLogger.trace(() -> MonitoringLogsDto.builder().level(LogLevel.INFO.getLevel()), "test {}", 1);
+
+            verify(logAppendFactory, times(1)).appendLog(any());
+        }
+
+        @Test
+        void testObsLogger() {
+            AtomicInteger counter = new AtomicInteger(0);
+            ObsLogger obsLogger = new ObsLogger() {
+                @Override
+                public MonitoringLogsDto.MonitoringLogsDtoBuilder logBaseBuilder() {
+                    return null;
+                }
+
+                @Override
+                public void trace(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+                    counter.incrementAndGet();
+                }
+
+                @Override
+                public void debug(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+
+                }
+
+                @Override
+                public void info(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+
+                }
+
+                @Override
+                public void warn(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, String message, Object... params) {
+
+                }
+
+                @Override
+                public void error(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, Throwable throwable, String message, Object... params) {
+
+                }
+
+                @Override
+                public void fatal(Callable<MonitoringLogsDto.MonitoringLogsDtoBuilder> callable, Throwable throwable, String message, Object... params) {
+
+                }
+
+                @Override
+                public boolean isEnabled(LogLevel logLevel) {
+                    return false;
+                }
+
+                @Override
+                public boolean isInfoEnabled() {
+                    return false;
+                }
+
+                @Override
+                public boolean isWarnEnabled() {
+                    return false;
+                }
+
+                @Override
+                public boolean isErrorEnabled() {
+                    return false;
+                }
+
+                @Override
+                public boolean isDebugEnabled() {
+                    return false;
+                }
+
+                @Override
+                public boolean isFatalEnabled() {
+                    return false;
+                }
+            };
+            obsLogger.trace("test {}", 1);
+            Assertions.assertEquals(1, counter.get());
+            obsLogger.trace("test {}", 2);
+            Assertions.assertEquals(2, counter.get());
+        }
+    }
+}
\ No newline at end of file
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/appender/FileAppenderTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/appender/FileAppenderTest.java
index 36f137576e..ef89928c02 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/appender/FileAppenderTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/logging/appender/FileAppenderTest.java
@@ -1,5 +1,8 @@
 package io.tapdata.observable.logging.appender;
 
+import com.tapdata.constant.BeanUtil;
+import com.tapdata.tm.commons.schema.MonitoringLogsDto;
+import io.tapdata.observable.logging.LogLevel;
 import io.tapdata.observable.logging.ObsLoggerFactory;
 import io.tapdata.observable.logging.util.Conf.LogConfiguration;
 import org.apache.logging.log4j.LogManager;
@@ -7,10 +10,15 @@
 import org.apache.logging.log4j.core.Appender;
 import org.apache.logging.log4j.core.appender.RollingFileAppender;
 import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.mockito.MockedStatic;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.*;
@@ -42,4 +50,68 @@ public void test1(){
             assertEquals("logs/jobs/test123.log", rollingFileAppender.getFileName());
         }
     }
+
+    @Nested
+    class testIncludeLevel {
+
+        private FileAppender fileAppender;
+        private Logger logger;
+
+        @BeforeEach
+        void before() {
+            try (MockedStatic<BeanUtil> mockStaticBeanUtil = mockStatic(BeanUtil.class)) {
+                mockStaticBeanUtil.when(() -> BeanUtil.getBean(any())).thenAnswer(answer -> {
+                    Class<?> cls = answer.getArgument(0);
+                    return mock(cls);
+                });
+                fileAppender = FileAppender.create("", "taskId");
+                Assertions.assertNotNull(fileAppender);
+            }
+
+            logger = mock(Logger.class);
+            ReflectionTestUtils.setField(fileAppender, "logger", logger);
+        }
+
+        @org.junit.jupiter.api.Test
+        void testOnlyDebug() {
+            fileAppender.include().include(LogLevel.DEBUG);
+
+            Object includeLevel = ReflectionTestUtils.getField(fileAppender, "includeLogLevel");
+            Assertions.assertNotNull(includeLevel);
+            Assertions.assertEquals(1, ((Set)includeLevel).size());
+
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.INFO.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.DEBUG.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.DEBUG.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.TRACE.getLevel()).build());
+
+            verify(logger, times(2)).debug(anyString());
+            verify(logger, times(0)).info(anyString());
+            verify(logger, times(0)).trace(anyString());
+        }
+
+        @org.junit.jupiter.api.Test
+        void testInclude() {
+            fileAppender.include().include(LogLevel.DEBUG).include(LogLevel.DEBUG).include(LogLevel.INFO);
+            Object includeLevel = ReflectionTestUtils.getField(fileAppender, "includeLogLevel");
+            Assertions.assertNotNull(includeLevel);
+            Assertions.assertEquals(2, ((Set)includeLevel).size());
+        }
+
+        @org.junit.jupiter.api.Test
+        void testTrace() {
+
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.TRACE.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.DEBUG.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.INFO.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.WARN.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.ERROR.getLevel()).build());
+            fileAppender.append(MonitoringLogsDto.builder().level(LogLevel.FATAL.getLevel()).build());
+
+
+            verify(logger, times(1)).debug(anyString());
+            verify(logger, times(1)).info(anyString());
+            verify(logger, times(1)).trace(anyString());
+        }
+    }
 }
\ No newline at end of file
diff --git a/iengine/modules/task-resource-supervisor-module/src/test/java/io/tapdata/supervisor/TaskResourceSupervisorManagerTest.java b/iengine/modules/task-resource-supervisor-module/src/test/java/io/tapdata/supervisor/TaskResourceSupervisorManagerTest.java
index 2e483b4bab..c3d36b4cfc 100644
--- a/iengine/modules/task-resource-supervisor-module/src/test/java/io/tapdata/supervisor/TaskResourceSupervisorManagerTest.java
+++ b/iengine/modules/task-resource-supervisor-module/src/test/java/io/tapdata/supervisor/TaskResourceSupervisorManagerTest.java
@@ -1,8 +1,13 @@
 package io.tapdata.supervisor;
 
 import com.tapdata.tm.commons.dag.Node;
+import io.tapdata.entity.script.ScriptFactory;
+import lombok.SneakyThrows;
 import org.junit.jupiter.api.*;
 import org.springframework.test.util.ReflectionTestUtils;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -16,8 +21,10 @@ public class TaskResourceSupervisorManagerTest {
     @Nested
     class CleanThreadGroupTest{
 
+        private TaskResourceSupervisorManager taskResourceSupervisorManager;
         @DisplayName("test cleanThreadGroup when destroy success ")
         @Test
+        @SneakyThrows
         void test1(){
             ThreadGroup threadGroup=new ThreadGroup("testThreadGroup");
             SupervisorAspectTask supervisorAspectTask=new SupervisorAspectTask();
@@ -30,15 +37,15 @@ void test1(){
             ReflectionTestUtils.setField(supervisorAspectTask,"threadGroupMap",threadGroupMap);
             taskNodeInfo.setSupervisorAspectTask(supervisorAspectTask);
 
-            TaskResourceSupervisorManager taskResourceSupervisorManager = new TaskResourceSupervisorManager();
-
-            ReflectionTestUtils.setField(taskResourceSupervisorManager,"CLEAN_LEAKED_THREAD_GROUP_THRESHOLD","0");
+            taskResourceSupervisorManager = new TaskResourceSupervisorManager();
+            reflectField("0");
             taskResourceSupervisorManager.addTaskSubscribeInfo(taskNodeInfo);
             taskResourceSupervisorManager.cleanThreadGroup();
             assertEquals(Boolean.FALSE,taskNodeInfo.isHasLaked());
         }
         @DisplayName("test cleanThreadGroup when destroy failed ")
         @Test
+        @SneakyThrows
         void test2(){
             ThreadGroup spyThreadGroup = new ThreadGroup("123");
             Thread[] threads = new Thread[1];
@@ -54,13 +61,21 @@ void test2(){
             taskNodeInfo.setHasLeaked(true);
             taskNodeInfo.setNodeThreadGroup(spyThreadGroup);
 
-            TaskResourceSupervisorManager taskResourceSupervisorManager = new TaskResourceSupervisorManager();
-            ReflectionTestUtils.setField(taskResourceSupervisorManager,"CLEAN_LEAKED_THREAD_GROUP_THRESHOLD","0");
+            taskResourceSupervisorManager = new TaskResourceSupervisorManager();
+            reflectField("0");
             taskResourceSupervisorManager.addTaskSubscribeInfo(taskNodeInfo);
             taskResourceSupervisorManager.cleanThreadGroup();
             atomicBoolean.set(true);
             assertEquals(Boolean.TRUE,taskNodeInfo.isHasLaked());
         }
+        private void reflectField(String threshold) throws NoSuchFieldException, IllegalAccessException {
+            Field scriptFactory = TaskResourceSupervisorManager.class.getDeclaredField("CLEAN_LEAKED_THREAD_GROUP_THRESHOLD");
+            scriptFactory.setAccessible(true);
+            Field modifiers = Field.class.getDeclaredField("modifiers");
+            modifiers.setAccessible(true);
+            modifiers.setInt(scriptFactory, scriptFactory.getModifiers() & ~Modifier.FINAL);
+            scriptFactory.set(taskResourceSupervisorManager, "0");
+        }
     }
     @Nested
     class TestSummary{
diff --git a/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
index 8cfdb73972..1888423eba 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
@@ -20,6 +20,8 @@
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
+import org.springframework.cache.Cache;
+import org.springframework.cache.caffeine.CaffeineCacheManager;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -29,6 +31,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Callable;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.*;
@@ -46,6 +49,8 @@ public class SettingsServiceTest {
     private AlarmMailService alarmMailService;
 
     private SettingsRepository mockSettingsRepository;
+
+    private CaffeineCacheManager caffeineCacheManager;
     @Nested
     class getMailAccountTest{
         @BeforeEach
@@ -55,8 +60,10 @@ void beforeEach(){
             mockSettingsRepository = mock(SettingsRepository.class);
             userService = mock(UserService.class);
             alarmMailService = mock(AlarmMailService.class);
+            caffeineCacheManager = mock(CaffeineCacheManager.class);
             settingsService.setMongoTemplate(mongoTemplate);
             settingsService.setSettingsRepository(mockSettingsRepository);
+            ReflectionTestUtils.setField(settingsService,"caffeineCacheManager",caffeineCacheManager);
             ReflectionTestUtils.setField(settingsService,"alarmMailService",alarmMailService);
             userDetail = new UserDetail("123", "customerId", "username", "password", "customerType",
                     "accessCode", false, false, false, false, Arrays.asList(new SimpleGrantedAuthority("role")));
@@ -164,8 +171,26 @@ void testFindALl_isDFS() {
             settings.setValue("CLOUD");
             List<Settings> list = new ArrayList<>();
             list.add(settings);
+            when(caffeineCacheManager.getCache("cloudSettings")).thenReturn(null);
+            when(mongoTemplate.find(any(Query.class), eq(Settings.class))).thenReturn(list);
+            final List<SettingsDto> result = settingsService.findALl("decode", filter);
+            assertThat(result.get(0).getValue()).isEqualTo(settings.getValue());
+        }
+
+        @Test
+        void testFindALl_isDFS_CACHE() {
+            final Filter filter = new Filter();
+            final Settings settings = new Settings();
+            settings.setKey("buildProfile");
+            settings.setValue("CLOUD");
+            List<Settings> list = new ArrayList<>();
+            list.add(settings);
+            Cache cache = mock(Cache.class);
+            when(cache.get(any(), any(Callable.class))).thenReturn(list);
+            when(caffeineCacheManager.getCache("cloudSettings")).thenReturn(cache);
             when(mongoTemplate.find(any(Query.class), eq(Settings.class))).thenReturn(list);
             final List<SettingsDto> result = settingsService.findALl("decode", filter);
+            verify(cache,times(1)).get(any(), any(Callable.class));
             assertThat(result.get(0).getValue()).isEqualTo(settings.getValue());
         }
 
diff --git a/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
index fa7958a5b0..e611ad2ab4 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
@@ -19,6 +19,7 @@
 import com.tapdata.tm.monitor.constant.KeyWords;
 import com.tapdata.tm.monitor.dto.TableSyncStaticDto;
 import com.tapdata.tm.monitor.entity.MeasurementEntity;
+import com.tapdata.tm.monitor.param.SyncStatusStatisticsParam;
 import com.tapdata.tm.monitor.vo.TableSyncStaticVo;
 import com.tapdata.tm.task.bean.TableStatusInfoDto;
 import com.tapdata.tm.task.service.TaskService;
@@ -792,4 +793,54 @@ void testQuerySyncStatic_snapshotRowTotal() {
         }
     }
 
+    @Nested
+    class cleanRemovedTableMeasurementTest {
+        @Test
+        void testCleanRemovedTableMeasurementNormal() {
+            String taskId = "111";
+            String taskRecordId = "222";
+            String tableName = "table1";
+            MongoTemplate mongoOperations = mock(MongoTemplate.class);
+            ReflectionTestUtils.setField(measurementServiceV2, "mongoOperations", mongoOperations);
+            doCallRealMethod().when(measurementServiceV2).cleanRemovedTableMeasurement(taskId, taskRecordId, tableName);
+            measurementServiceV2.cleanRemovedTableMeasurement(taskId, taskRecordId, tableName);
+            verify(mongoOperations, new Times(1)).remove(any(Query.class), any(Class.class), anyString());
+        }
+    }
+
+    @Nested
+    class findRunTableTest {
+        @Test
+        void testFindRunTableNormal() {
+            String taskId = "111";
+            String taskRecordId = "222";
+            MongoTemplate mongoOperations = mock(MongoTemplate.class);
+            ReflectionTestUtils.setField(measurementServiceV2, "mongoOperations", mongoOperations);
+            List<MeasurementEntity> measurementEntities = new ArrayList<>();
+            MeasurementEntity measurementEntity = mock(MeasurementEntity.class);
+            measurementEntities.add(measurementEntity);
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table", "test");
+            when(measurementEntity.getTags()).thenReturn(tags);
+            when(mongoOperations.find(any(Query.class), any(Class.class), anyString())).thenReturn(measurementEntities);
+            doCallRealMethod().when(measurementServiceV2).findRunTable(taskId, taskRecordId);
+            List<String> actual = measurementServiceV2.findRunTable(taskId, taskRecordId);
+            verify(mongoOperations, new Times(1)).find(any(Query.class), any(Class.class), anyString());
+            assertEquals("test", actual.get(0));
+        }
+    }
+
+    @Nested
+    class queryTableSyncStatusStatisticsTest{
+        @Test
+        void testQueryTableSyncStatusStatisticsNormal(){
+            MongoTemplate mongoOperations = mock(MongoTemplate.class);
+            ReflectionTestUtils.setField(measurementServiceV2, "mongoOperations", mongoOperations);
+            when(mongoOperations.aggregate(any(Aggregation.class), anyString(), any(Class.class))).thenReturn(mock(AggregationResults.class));
+            SyncStatusStatisticsParam param = mock(SyncStatusStatisticsParam.class);
+            doCallRealMethod().when(measurementServiceV2).queryTableSyncStatusStatistics(param);
+            measurementServiceV2.queryTableSyncStatusStatistics(param);
+            verify(mongoOperations, new Times(1)).aggregate(any(Aggregation.class), anyString(), any(Class.class));
+        }
+    }
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImplTest.java
new file mode 100644
index 0000000000..e549927d78
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/monitoringlogs/service/MonitoringLogsServiceImplTest.java
@@ -0,0 +1,57 @@
+package com.tapdata.tm.monitoringlogs.service;
+
+import com.tapdata.tm.base.dto.Page;
+import com.tapdata.tm.commons.schema.MonitoringLogsDto;
+import com.tapdata.tm.monitoringlogs.entity.MonitoringLogsEntity;
+import com.tapdata.tm.monitoringlogs.param.MonitoringLogQueryParam;
+import com.tapdata.tm.monitoringlogs.repository.MonitoringLogsRepository;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Query;
+
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+
+/**
+ * @author lg&lt;lirufei0808@gmail.com&gt;
+ * create at 2024/11/20 14:02
+ */
+public class MonitoringLogsServiceImplTest {
+
+    @Test
+    void testQuery() throws ExecutionException, InterruptedException {
+
+        MonitoringLogsRepository monitoringLogsRepository = mock(MonitoringLogsRepository.class);
+        MongoTemplate mongoOperations = mock(MongoTemplate.class);
+
+        when(mongoOperations.count(any(Query.class), eq(MonitoringLogsEntity.class))).thenReturn(0L);
+
+        MonitoringLogsServiceImpl monitoringLogsService =
+                new MonitoringLogsServiceImpl(monitoringLogsRepository, CompletableFuture.completedFuture(mongoOperations));
+
+        MonitoringLogQueryParam param = new MonitoringLogQueryParam();
+        param.setTaskId("673d825967ac3c7c5785ae02");
+        param.setStart(1732421532000L);
+        param.setType("testRun");
+        param.setOrder("asc");
+
+        param.setIncludeLogTags(Arrays.asList("tag1", "tag2"));
+
+        Page<MonitoringLogsDto> page = monitoringLogsService.query(param);
+
+
+        param.setIncludeLogTags(null);
+        param.setExcludeLogTags(Arrays.asList("tag1", "tag2"));
+
+        page = monitoringLogsService.query(param);
+
+        verify(mongoOperations, times(2)).count(any(Query.class), eq(MonitoringLogsEntity.class));
+
+    }
+
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
index 7def440231..19ef2c9f8a 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
@@ -5323,4 +5323,44 @@ void testNewConnIsNull() {
 
         }
     }
+
+    @Nested
+    class cleanRemovedTableMeasurementIfNeedTest {
+        private TaskDto taskDto;
+        private MeasurementServiceV2 measurementServiceV2;
+        private String taskId;
+        private String taskRecordId;
+
+        @BeforeEach
+        void beforeEach() {
+            taskDto = mock(TaskDto.class);
+            taskId = "6739ff5046a4de5089e231de";
+            when(taskDto.getId()).thenReturn(new ObjectId(taskId));
+            taskRecordId = "673a010846a4de5089e278d9";
+            when(taskDto.getTaskRecordId()).thenReturn(taskRecordId);
+            measurementServiceV2 = mock(MeasurementServiceV2.class);
+            ReflectionTestUtils.setField(taskService, "measurementServiceV2", measurementServiceV2);
+        }
+
+        @Test
+        @DisplayName("test cleanRemovedTableMeasurementIfNeed normal")
+        void test1() {
+            DAG dag = mock(DAG.class);
+            when(taskDto.getDag()).thenReturn(dag);
+            LinkedList<DatabaseNode> sourceNode = new LinkedList<>();
+            DatabaseNode node = mock(DatabaseNode.class);
+            sourceNode.add(node);
+            when(dag.getSourceNode()).thenReturn(sourceNode);
+            List<String> tables = new ArrayList<>();
+            tables.add("table1");
+            when(node.getTableNames()).thenReturn(tables);
+            List<String> runTables = new ArrayList<>();
+            runTables.add("table1");
+            runTables.add("table2");
+            when(measurementServiceV2.findRunTable(anyString(), anyString())).thenReturn(runTables);
+            doCallRealMethod().when(taskService).cleanRemovedTableMeasurementIfNeed(taskDto);
+            taskService.cleanRemovedTableMeasurementIfNeed(taskDto);
+            verify(measurementServiceV2, new Times(1)).cleanRemovedTableMeasurement(taskId, taskRecordId, "table2");
+        }
+    }
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
index e51c14a759..e8b9adc645 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/LdpServiceImplTest.java
@@ -9,6 +9,7 @@
 import com.tapdata.tm.commons.dag.nodes.TableNode;
 import com.tapdata.tm.commons.dag.vo.SyncObjects;
 import com.tapdata.tm.commons.schema.DataSourceConnectionDto;
+import com.tapdata.tm.commons.schema.Field;
 import com.tapdata.tm.commons.schema.MetadataInstancesDto;
 import com.tapdata.tm.commons.schema.Tag;
 import com.tapdata.tm.commons.schema.bean.SourceDto;
@@ -344,9 +345,9 @@ void beforeEach() {
         void testAfterLdpTaskSimple() {
             user = mock(UserDetail.class);
             String taskId = "6720c4a18c6b586b9e1b493b";
-            when(taskService.findById(any(ObjectId.class))).thenReturn(mock(TaskDto.class));
+            when(taskService.findByTaskId(any(),any())).thenReturn(mock(TaskDto.class));
             doCallRealMethod().when(ldpService).afterLdpTask(taskId, user);
             ldpService.afterLdpTask(taskId, user);
         }
     }
-}
\ No newline at end of file
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/user/service/UserServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/user/service/UserServiceImplTest.java
index 1a68b78b0c..43d13b29a0 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/user/service/UserServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/user/service/UserServiceImplTest.java
@@ -1,5 +1,7 @@
 package com.tapdata.tm.user.service;
 
+import com.tapdata.tm.Permission.dto.PermissionDto;
+import com.tapdata.tm.Permission.service.PermissionService;
 import com.tapdata.tm.Settings.constant.CategoryEnum;
 import com.tapdata.tm.Settings.constant.KeyEnum;
 import com.tapdata.tm.Settings.constant.SettingUtil;
@@ -8,11 +10,15 @@
 import com.tapdata.tm.Settings.service.SettingsService;
 import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.config.security.UserDetail;
+import com.tapdata.tm.role.dto.RoleDto;
+import com.tapdata.tm.role.service.RoleService;
 import com.tapdata.tm.roleMapping.dto.RoleMappingDto;
 import com.tapdata.tm.roleMapping.service.RoleMappingService;
 import com.tapdata.tm.user.dto.LdapLoginDto;
 import com.tapdata.tm.user.dto.TestLdapDto;
+import com.tapdata.tm.user.dto.UserDto;
 import lombok.SneakyThrows;
+import org.bson.types.ObjectId;
 import org.junit.jupiter.api.*;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
@@ -30,6 +36,7 @@
 import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -39,14 +46,20 @@ public class UserServiceImplTest {
     private UserServiceImpl userService;
     private RoleMappingService roleMappingService;
     private SettingsService settingsService;
+    private RoleService roleService;
+    private PermissionService permissionService;
 
     @BeforeEach
     void beforeEach(){
         userService = mock(UserServiceImpl.class);
         roleMappingService = mock(RoleMappingService.class);
         settingsService = mock(SettingsService.class);
+        roleService = mock(RoleService.class);
+        permissionService = mock(PermissionService.class);
         ReflectionTestUtils.setField(userService, "settingsService", settingsService);
         ReflectionTestUtils.setField(userService, "roleMappingService", roleMappingService);
+        ReflectionTestUtils.setField(userService, "roleService", roleService);
+        ReflectionTestUtils.setField(userService, "permissionService", permissionService);
     }
 
     @Nested
@@ -487,6 +500,25 @@ public void testBuildDirContext_SSL() {
             when(sslContext.getSocketFactory()).thenReturn(mock(SSLSocketFactory.class));
             assertThrows(NamingException.class, () -> userService.buildDirContext(adLoginDto));
         }
+
+        @Test
+        @SneakyThrows
+        public void testBuildDirContext_BindDn_Contains() {
+            String baseDN = "dc=example,dc=com";
+            LdapLoginDto adLoginDto = LdapLoginDto.builder()
+                    .ldapUrl("ldap://example.com:389")
+                    .baseDN(baseDN)
+                    .bindDN("admin")
+                    .password("password")
+                    .sslEnable(true)
+                    .cert(certString)
+                    .build();
+            SSLContext sslContext = mock(SSLContext.class);
+            doCallRealMethod().when(userService).convertBaseDnToDomain(baseDN);
+            when(userService.createSSLContext(any(InputStream.class))).thenReturn(sslContext);
+            when(sslContext.getSocketFactory()).thenReturn(mock(SSLSocketFactory.class));
+            assertThrows(NamingException.class, () -> userService.buildDirContext(adLoginDto));
+        }
     }
 
     @Nested
@@ -541,4 +573,74 @@ public void testCreateSSLContext() throws Exception {
             "hinNPHvFAMDUY7HfrdDfCoEIZSK4u/xFQscsNsE3hywVYYN7uz8wy502fEHzwIgp\n" +
             "/3vC2igEWcqhxJn9iiIHAzA0mZoS1DOKhgFG\n" +
             "-----END CERTIFICATE-----";
+
+
+    @Nested
+    class GetUserDetailTest{
+
+        @Test
+        void test_main(){
+            doCallRealMethod().when(userService).getUserDetail(anyString());
+            UserDto userDto = new UserDto();
+            userDto.setCreateAt(new Date());
+            when(userService.findById(any())).thenReturn(userDto);
+            List<RoleMappingDto> roleMappingDtoList =  new ArrayList<>();
+            RoleMappingDto roleMappingDto = new RoleMappingDto();
+            ObjectId id = new ObjectId();
+            roleMappingDto.setRoleId(id);
+            roleMappingDtoList.add(roleMappingDto);
+            List<RoleDto> roleDtos = new ArrayList<>();
+            RoleDto roleDto = new RoleDto();
+            roleDto.setId(id);
+            roleDtos.add(roleDto);
+            when(roleService.findAll(any(Query.class))).thenReturn(roleDtos);
+            when(roleMappingService.getUser(any(),any())).thenReturn(roleMappingDtoList);
+            List<PermissionDto> permissionDtos = new ArrayList<>();
+            PermissionDto permissionDto = new PermissionDto();
+            permissionDto.setId("test");
+            permissionDtos.add(permissionDto);
+            when(permissionService.getCurrentPermission(anyString())).thenReturn(permissionDtos);
+            UserDto result = userService.getUserDetail("test");
+            Assertions.assertEquals(1,result.getPermissions().size());
+            Assertions.assertEquals(1,result.getRoleMappings().size());
+        }
+
+        @Test
+        void test_roleDtosIsNull(){
+            doCallRealMethod().when(userService).getUserDetail(anyString());
+            UserDto userDto = new UserDto();
+            userDto.setCreateAt(new Date());
+            when(userService.findById(any())).thenReturn(userDto);
+            List<RoleMappingDto> roleMappingDtoList =  new ArrayList<>();
+            RoleMappingDto roleMappingDto = new RoleMappingDto();
+            ObjectId id = new ObjectId();
+            roleMappingDto.setRoleId(id);
+            roleMappingDtoList.add(roleMappingDto);
+            when(roleService.findAll(any(Query.class))).thenReturn(new ArrayList<>());
+            when(roleMappingService.getUser(any(),any())).thenReturn(roleMappingDtoList);
+            List<PermissionDto> permissionDtos = new ArrayList<>();
+            PermissionDto permissionDto = new PermissionDto();
+            permissionDto.setId("test");
+            permissionDtos.add(permissionDto);
+            when(permissionService.getCurrentPermission(anyString())).thenReturn(permissionDtos);
+            UserDto result = userService.getUserDetail("test");
+            Assertions.assertEquals(1,result.getPermissions().size());
+            Assertions.assertEquals(1,result.getRoleMappings().size());
+        }
+
+        @Test
+        void test_roleMappingAndPermissionIsEmpty(){
+            doCallRealMethod().when(userService).getUserDetail(anyString());
+            UserDto userDto = new UserDto();
+            userDto.setCreateAt(new Date());
+            when(userService.findById(any())).thenReturn(userDto);
+            when(roleMappingService.getUser(any(),any())).thenReturn(new ArrayList<>());
+            when(permissionService.getCurrentPermission(anyString())).thenReturn(new ArrayList<>());
+            UserDto result = userService.getUserDetail("test");
+            Assertions.assertNull(result.getRoleMappings());
+        }
+
+
+    }
 }
+