(dolphinscheduler) branch dev updated: [Fix-18543][DAO] Return only the valid task instance per task code (#18544)

[zihaoxiang]

This is an automated email from the ASF dual-hosted git repository.

SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git


The following commit(s) were added to refs/heads/dev by this push:
     new 9839c418c1 [Fix-18543][DAO] Return only the valid task instance per 
task code (#18544)
9839c418c1 is described below

commit 9839c418c1a6d7f2a8c3395552edcd99c2de1b37
Author: Sepuri Sai Krishna <saik20...@gmail.com>
AuthorDate: Tue Sep 15 08:26:18 2026 +0530

    [Fix-18543][DAO] Return only the valid task instance per task code (#18544)
---
 .../dao/mapper/TaskInstanceMapper.xml              |  2 ++
 .../repository/impl/TaskInstanceDaoImplTest.java   | 28 +++++++++++++++++++++-
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git 
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
 
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
index e632e7c9b3..de1883346a 100644
--- 
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
+++ 
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
@@ -258,6 +258,7 @@
         where 1=1
         and workflow_instance_id = #{workflowInstanceId}
         and state != 8
+        and flag = 1
         <if test="taskCodes != null and taskCodes.size() != 0">
             and task_code in
             <foreach collection="taskCodes" index="index" item="i" open="(" 
separator="," close=")">
@@ -269,6 +270,7 @@
         on instance.workflow_instance_id = t_max.workflow_instance_id
         and instance.task_code = t_max.task_code
         and instance.end_time = t_max.max_end_time
+        and instance.flag = 1
     </select>
     <select id="findLastTaskInstance" 
resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
         select
diff --git 
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/TaskInstanceDaoImplTest.java
 
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/TaskInstanceDaoImplTest.java
index 226409b197..e9e85c4848 100644
--- 
a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/TaskInstanceDaoImplTest.java
+++ 
b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/TaskInstanceDaoImplTest.java
@@ -27,6 +27,7 @@ import 
org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
@@ -65,6 +66,27 @@ class TaskInstanceDaoImplTest extends BaseDaoTest {
         assertEquals(later.getTime() / 1000, 
extractResult.getEndTime().getTime() / 1000);
     }
 
+    @Test
+    void 
queryLastTaskInstanceListIntervalInWorkflowInstanceWhenAttemptsShareTheSameEndTime()
 {
+        Date sameEndTime = new Date();
+
+        // A failed attempt and the retry which replaced it, both finishing 
within the same second. On MySQL
+        // t_ds_task_instance.end_time is a datetime without fractional 
seconds, so the two attempts end up
+        // with exactly the same end_time. This is exactly the state 
RetryTaskInstanceFactory leaves behind:
+        // the superseded attempt is flagged invalid and the new attempt is 
the only valid one.
+        insertTaskInstance(EXTRACT_TASK, TaskExecutionStatus.FAILURE, 
sameEndTime, Flag.NO);
+        insertTaskInstance(EXTRACT_TASK, TaskExecutionStatus.SUCCESS, 
sameEndTime, Flag.YES);
+
+        Set<Long> taskCodes = new 
HashSet<>(Collections.singletonList(EXTRACT_TASK));
+        List<TaskInstance> result = 
taskInstanceDao.queryLastTaskInstanceListIntervalInWorkflowInstance(
+                WORKFLOW_INSTANCE_ID, taskCodes);
+
+        // Only the valid attempt should be returned, otherwise callers which 
key the result by taskCode,
+        // e.g. DependentExecute, fail with "IllegalStateException: Duplicate 
key".
+        assertEquals(1, result.size());
+        assertEquals(TaskExecutionStatus.SUCCESS, result.get(0).getState());
+    }
+
     @Test
     void queryLastTaskInstanceIntervalInWorkflowInstance() {
         Date earlier = new Date(System.currentTimeMillis() - 3600_000);
@@ -80,6 +102,10 @@ class TaskInstanceDaoImplTest extends BaseDaoTest {
     }
 
     private void insertTaskInstance(long taskCode, TaskExecutionStatus state, 
Date endTime) {
+        insertTaskInstance(taskCode, state, endTime, Flag.YES);
+    }
+
+    private void insertTaskInstance(long taskCode, TaskExecutionStatus state, 
Date endTime, Flag flag) {
         TaskInstance ti = TaskInstance.builder()
                 .name("shell-task-" + taskCode)
                 .taskType("SHELL")
@@ -88,7 +114,7 @@ class TaskInstanceDaoImplTest extends BaseDaoTest {
                 .taskCode(taskCode)
                 .taskDefinitionVersion(1)
                 .state(state)
-                .flag(Flag.YES)
+                .flag(flag)
                 .submitTime(new Date())
                 .firstSubmitTime(new Date())
                 .startTime(new Date())