Repository: tapdata/tapdata (language: python)

Issue:
fix TAP-2090 regard time bug



Discussion / hints:

Fix patch:
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 14d2e9606d..aa464bb919 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
@@ -328,9 +328,19 @@ public interface TaskMergeProcessorExCode_16 {
 			describeCN = "清理并销毁缓存失败"
 	)
 	String CLEAR_AND_DESTROY_CACHE_FAILED = "16058";
+	@TapExCode(
+			describe = "Join key value convert number failed",
+			describeCN = "关联键值转换为数字失败"
+	)
+	String JOIN_KEY_VALUE_CONVERT_NUMBER_FAILED = "16059";
+	@TapExCode(
+			describe = "Look up cache data, unknown error occurred",
+			describeCN = "反查缓存数据，发生未知错误"
+	)
+	String LOOK_UP_UNKNOWN_ERROR = "16060";
 	@TapExCode(
 			describe = "Lookup CompletableFuture list cannot be null",
 			describeCN = "反查数据的CompletableFuture列表不能为空"
 	)
-	String LOOKUP_COMPLETABLE_FUTURE_LIST_IS_NULL = "16059";
+	String LOOKUP_COMPLETABLE_FUTURE_LIST_IS_NULL = "16061";
 }
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 7c43061f32..6f6c49bff7 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
@@ -1192,7 +1192,12 @@ private TapAdvanceFilter batchFilterRead() {
 					TapType tapType = tapField.getTapType();
 					Object convertValue;
 					if (queryOperator.getValue() != null) {
-						convertValue = getConvertValue(tapType, queryOperator.getValue().toString());
+						if(queryOperator.getOriginalValue() == null) {
+							convertValue = getConvertValue(tapType, queryOperator.getValue().toString());
+							queryOperator.setOriginalValue(queryOperator.getValue().toString());
+						}else {
+							convertValue = getConvertValue(tapType, queryOperator.getOriginalValue().toString());
+						}
 						queryOperator.setValue(convertValue);
 					}
 					if (EQUAL_VALUE == queryOperator.getOperator()) {
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 d3fff8a54c..4423629f0d 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
@@ -2,10 +2,7 @@
 
 import com.hazelcast.core.HazelcastInstance;
 import com.tapdata.constant.*;
-import com.tapdata.entity.Connections;
-import com.tapdata.entity.OperationType;
-import com.tapdata.entity.SyncStage;
-import com.tapdata.entity.TapdataEvent;
+import com.tapdata.entity.*;
 import com.tapdata.entity.task.context.DataProcessorContext;
 import com.tapdata.tm.commons.dag.Edge;
 import com.tapdata.tm.commons.dag.Node;
@@ -51,6 +48,8 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.bson.BsonType;
 import org.bson.Document;
 import org.jetbrains.annotations.NotNull;
@@ -88,6 +87,7 @@ public class HazelcastMergeNode extends HazelcastProcessorBaseNode implements Me
 	public static final String UPDATE_JOIN_KEY_VALUE_CACHE_IN_MEM_SIZE_PROP_KEY = "UPDATE_JOIN_KEY_VALUE_CACHE_IN_MEM_SIZE";
 	public static final String HANDLE_UPDATE_JOIN_KEY_THREAD_NUM_PROP_KEY = "HANDLE_UPDATE_JOIN_KEY_THREAD_NUM";
 	public static final int DEFAULT_UPDATE_JOIN_KEY_THREAD_NUM = 4;
+	private Logger logger = LogManager.getLogger(HazelcastMergeNode.class);
 
 	// 缓存表信息{"前置节点id": "Hazelcast缓存资源{"join value string": {"pk value string": "after data"}}"}
 	private Map<String, ConstructIMap<Document>> mergeCacheMap;
@@ -1302,10 +1302,11 @@ private String getJoinValueKeyBySource(Map<String, Object> data, MergeTablePrope
 			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));
 			}
-			if (value instanceof Number) {
-				values.add(new BigDecimal(String.valueOf(value)).stripTrailingZeros().toPlainString());
-			} else {
-				values.add(String.valueOf(value));
+			try {
+				values.add(convertJoinKeyValue2String(value));
+			} catch (JoinKeyValueConvertNumberException e) {
+				throw new TapCodeException(TaskMergeProcessorExCode_16.JOIN_KEY_VALUE_CONVERT_NUMBER_FAILED, String.format("- Merge table: %s%n- Map name: %s%n- Join key: %s%n- Data: %s(%s)",
+						mergeProperty.getTableName(), hazelcastConstruct.getName(), joinKey, value, null == value ? "null" : value.getClass().getName()));
 			}
 		}
 		return String.join("_", values);
@@ -1329,15 +1330,40 @@ private String getJoinValueKeyByTarget(Map<String, Object> data, MergeTablePrope
 			if (value instanceof NotExistsNode) {
 				return null;
 			}
-			if (value instanceof Number) {
-				values.add(new BigDecimal(String.valueOf(value)).stripTrailingZeros().toPlainString());
-			} else {
-				values.add(String.valueOf(value));
+			try {
+				values.add(convertJoinKeyValue2String(value));
+			} catch (JoinKeyValueConvertNumberException e) {
+				throw new TapCodeException(TaskMergeProcessorExCode_16.JOIN_KEY_VALUE_CONVERT_NUMBER_FAILED, String.format("- Merge table: %s%n- Map name: %s%n- Join key: %s%n- Data: %s(%s)",
+						mergeProperty.getTableName(), hazelcastConstruct.getName(), joinKey, value, null == value ? "null" : value.getClass().getName()), e);
 			}
 		}
 		return String.join("_", values);
 	}
 
+	private static String convertJoinKeyValue2String(Object value) throws JoinKeyValueConvertNumberException {
+		if (null == value) {
+			return "";
+		}
+		if (value instanceof Number) {
+			try {
+				if (value instanceof Double && Double.isNaN((Double) value)) {
+					return "";
+				}
+				return new BigDecimal(String.valueOf(value)).stripTrailingZeros().toPlainString();
+			} catch (Exception e) {
+				throw new JoinKeyValueConvertNumberException(e);
+			}
+		} else {
+			return String.valueOf(value);
+		}
+	}
+
+	private static class JoinKeyValueConvertNumberException extends Exception {
+		public JoinKeyValueConvertNumberException(Throwable cause) {
+			super(cause);
+		}
+	}
+
 	private String encode(String str) {
 		return Base64.getEncoder().encodeToString(str.getBytes(StandardCharsets.UTF_8));
 	}
@@ -1441,12 +1467,10 @@ protected List<MergeLookupResult> lookup(TapdataEvent tapdataEvent) {
 		List<MergeLookupResult> mergeLookupResults;
 		try {
 			mergeLookupResults = recursiveLookup(currentMergeTableProperty, after, true);
+		} catch (TapCodeException e) {
+			throw new TapEventException(e.getCode(), e.getMessage(), e).addEvent(tapdataEvent.getTapEvent());
 		} catch (Exception e) {
-			if (e instanceof TapCodeException) {
-				throw new TapEventException(((TapCodeException) e).getCode(), e.getMessage(), e.getCause()).addEvent(tapdataEvent.getTapEvent());
-			} else {
-				throw e;
-			}
+			throw new TapEventException(TaskMergeProcessorExCode_16.LOOK_UP_UNKNOWN_ERROR, e).addEvent(tapdataEvent.getTapEvent());
 		}
 		return mergeLookupResults;
 	}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/agent/service/AgentGroupService.java b/manager/tm/src/main/java/com/tapdata/tm/agent/service/AgentGroupService.java
index 0fc3287cd9..f2dcd260c2 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/agent/service/AgentGroupService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/agent/service/AgentGroupService.java
@@ -215,9 +215,20 @@ protected List<AgentGroupDto> updateAgent(List<String> groupIds, List<String> ne
         if (CollectionUtils.isEmpty(groupIds)) {
             return Lists.newArrayList();
         }
-        Criteria criteria = findCriteria(groupIds);
-        update(Query.query(criteria),
-                new Update().set(AgentGroupTag.TAG_AGENT_IDS, newAgentIds), loginUser);
+        List<AgentGroupEntity> all = findAll(Query.query(Criteria.where("agentIds").in(newAgentIds)), loginUser);
+        List<String> cleanAgentIds = new ArrayList<>();
+        all.forEach(a -> {
+            String groupId = a.getGroupId();
+            if (!groupIds.contains(groupId)) {
+                cleanAgentIds.add(groupId);
+            }
+        });
+
+        update(Query.query(findCriteria(groupIds)), new Update().set(AgentGroupTag.TAG_AGENT_IDS, newAgentIds), loginUser);
+        if (!cleanAgentIds.isEmpty()) {
+            update(Query.query(findCriteria(cleanAgentIds)), new Update().set(AgentGroupTag.TAG_AGENT_IDS, new ArrayList<>()), loginUser);
+            groupIds.addAll(cleanAgentIds);
+        }
         return findAgentGroupInfoMany(groupIds, loginUser);
     }
 


Test patch:
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
index b3eaf12cb6..c8f2641cdc 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
@@ -3,6 +3,7 @@
 import base.hazelcast.BaseHazelcastNodeTest;
 import com.tapdata.entity.dataflow.SyncProgress;
 import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.TableNode;
 import io.tapdata.entity.codec.filter.TapCodecsFilterManager;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.ddl.TapDDLEvent;
@@ -11,18 +12,27 @@
 import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import io.tapdata.entity.schema.TapField;
+import io.tapdata.entity.schema.TapTable;
+import io.tapdata.entity.schema.type.TapDateTime;
+import io.tapdata.entity.schema.type.TapType;
+import io.tapdata.entity.schema.value.DateTime;
+import io.tapdata.pdk.apis.entity.QueryOperator;
 import io.tapdata.pdk.core.api.ConnectorNode;
+import io.tapdata.schema.TapTableMap;
 import lombok.SneakyThrows;
+import org.junit.Assert;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
@@ -180,4 +190,58 @@ void testIsNotPollingCDC() {
 			verify(tapInsertRecordEvent, times(0)).getAfter();
 		}
 	}
+
+	@Test
+	@DisplayName("test QueryOperator ")
+	void testQueryOperator() {
+		String dateTime = "2023-12-20 12:23:20";
+		QueryOperator queryOperator = new QueryOperator("createTime", dateTime, 1);
+		queryOperator(dateTime, queryOperator);
+		Assert.assertTrue(queryOperator.getOriginalValue() == dateTime);
+	}
+
+
+	@Test
+	@DisplayName("test QueryOperator HasOriginalValue ")
+	void testQueryOperatorHasOriginalValue() {
+		String dateTime = "2021-12-20 12:23:24";
+		QueryOperator queryOperator = new QueryOperator("createTime", dateTime, 1);
+		queryOperator.setOriginalValue(dateTime);
+		queryOperator(dateTime, queryOperator);
+		LocalDateTime localDateTime;
+		String datetimeFormat = "yyyy-MM-dd HH:mm:ss";
+		try {
+			localDateTime = LocalDateTime.parse(dateTime, DateTimeFormatter.ofPattern(datetimeFormat));
+		} catch (Exception e) {
+			throw new RuntimeException("The input string format is incorrect, expected format: " + datetimeFormat + ", actual value: " + dateTime);
+		}
+		ZonedDateTime gmtZonedDateTime = localDateTime.atZone(ZoneId.of("GMT"));
+		DateTime expectedValue = new DateTime(gmtZonedDateTime);
+		Assert.assertTrue(expectedValue.compareTo((DateTime) queryOperator.getValue()) == 0);
+	}
+
+
+	public void queryOperator(String dateTime,QueryOperator queryOperator){
+		List<QueryOperator> conditions = new ArrayList<>();
+
+		conditions.add(queryOperator);
+		TableNode tableNodeTemp =new TableNode();
+		tableNodeTemp.setIsFilter(true);
+		tableNodeTemp.setConditions(conditions);
+		tableNodeTemp.setTableName("test");
+		when(dataProcessorContext.getNode()).thenReturn((Node) tableNodeTemp);
+		TapTable tapTable = new TapTable();
+		LinkedHashMap<String, TapField> nameFieldMap = new LinkedHashMap<>();
+		TapField tapField = new TapField("createTime", "TapDateTime");
+		tapField.setTapType(new TapDateTime());
+		nameFieldMap.put("createTime",tapField);
+		tapTable.setNameFieldMap(nameFieldMap);
+		tapTable.setName("test");
+		tapTable.setId("test");
+		TapTableMap<String, TapTable>  tapTableMap =TapTableMap.create("test",tapTable);
+
+		when(dataProcessorContext.getTapTableMap()).thenReturn(tapTableMap);
+		ReflectionTestUtils.invokeMethod(hazelcastSourcePdkDataNode,"batchFilterRead");
+
+	}
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/agent/service/AgentGroupServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/agent/service/AgentGroupServiceTest.java
index dc703c031b..6cdd50e726 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/agent/service/AgentGroupServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/agent/service/AgentGroupServiceTest.java
@@ -23,6 +23,7 @@
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.ds.service.impl.DataSourceService;
 import com.tapdata.tm.task.service.TaskServiceImpl;
+import com.tapdata.tm.user.entity.User;
 import com.tapdata.tm.utils.Lists;
 import com.tapdata.tm.worker.dto.WorkerDto;
 import com.tapdata.tm.worker.service.WorkerServiceImpl;
@@ -572,12 +573,24 @@ class UpdateAgentTest {
         List<String> groupIds;
         List<String> agentId;
         Criteria criteria;
+
+        List<AgentGroupEntity> all;
+        AgentGroupEntity entity;
         @BeforeEach
         void init() {
             agentId = mock(List.class);
             groupIds = mock(List.class);
             when(groupIds.isEmpty()).thenReturn(false);
 
+            all = new ArrayList<>();
+            entity = mock(AgentGroupEntity.class);
+            when(entity.getGroupId()).thenReturn("id");
+            all.add(entity);
+            when(groupIds.contains("id")).thenReturn(true);
+            when(groupIds.addAll(anyList())).thenReturn(true);
+            when(agentGroupService.findAll(any(Query.class), any(UserDetail.class))).thenReturn(all);
+
+
             criteria = mock(Criteria.class);
             when(agentGroupService.findCriteria(anyList())).thenReturn(criteria);
             when(agentGroupService.update(any(Query.class), any(Update.class), any(UserDetail.class))).thenReturn(mock(UpdateResult.class));
@@ -587,11 +600,9 @@ void init() {
 
         @Test
         void testNormal() {
-            try(MockedStatic<Query> q = mockStatic(Query.class)) {
-                q.when(() -> Query.query(criteria)).thenReturn(mock(Query.class));
-                Assertions.assertDoesNotThrow(() -> agentGroupService.updateAgent(groupIds, agentId, userDetail));
-                q.verify(() -> Query.query(criteria), times(1));
-            }
+            Assertions.assertDoesNotThrow(() -> agentGroupService.updateAgent(groupIds, agentId, userDetail));
+            verify(groupIds, times(0)).addAll(anyList());
+            verify(groupIds, times(1)).contains("id");
             verify(groupIds, times(1)).isEmpty();
             verify(agentGroupService, times(1)).findCriteria(groupIds);
             verify(agentGroupService, times(1)).update(any(Query.class), any(Update.class), any(UserDetail.class));
@@ -601,16 +612,26 @@ void testNormal() {
         @Test
         void testGroupIdsIsEmpty() {
             when(groupIds.isEmpty()).thenReturn(true);
-            try(MockedStatic<Query> q = mockStatic(Query.class)) {
-                q.when(() -> Query.query(criteria)).thenReturn(mock(Query.class));
-                Assertions.assertDoesNotThrow(() -> agentGroupService.updateAgent(groupIds, agentId, userDetail));
-                q.verify(() -> Query.query(criteria), times(0));
-            }
+            Assertions.assertDoesNotThrow(() -> agentGroupService.updateAgent(groupIds, agentId, userDetail));
+            verify(groupIds, times(0)).addAll(anyList());
+            verify(groupIds, times(0)).contains("id");
             verify(groupIds, times(1)).isEmpty();
             verify(agentGroupService, times(0)).findCriteria(groupIds);
             verify(agentGroupService, times(0)).update(any(Query.class), any(Update.class), any(UserDetail.class));
             verify(agentGroupService, times(0)).findAgentGroupInfoMany(groupIds, userDetail);
         }
+
+        @Test
+        void testNeedClean() {
+            when(groupIds.contains("id")).thenReturn(false);
+            Assertions.assertDoesNotThrow(() -> agentGroupService.updateAgent(groupIds, agentId, userDetail));
+            verify(groupIds, times(1)).addAll(anyList());
+            verify(groupIds, times(1)).isEmpty();
+            verify(agentGroupService, times(1)).findCriteria(groupIds);
+            verify(agentGroupService, times(2)).update(any(Query.class), any(Update.class), any(UserDetail.class));
+            verify(agentGroupService, times(1)).findAgentGroupInfoMany(groupIds, userDetail);
+            verify(groupIds, times(1)).contains("id");
+        }
     }
 
     @Nested