[Fix-18543][DAO] Return only the valid task instance per task code

[SEPURI-SAI-KRISHNA]

Was this PR generated or assisted by AI?

YES. The bug was found, and the fix and the test were written with AI assistance
(Claude Code). I reviewed the change at a high level and verified it via the added test
and the module test suite.

Purpose of the pull request

Closes #18543.

findLastTaskInstances resolves the last task instance per task code by grouping on
task_code with max(end_time) and joining back on instance.end_time = t_max.max_end_time.
Neither side filters on flag.

Retry, failover and failed-recover keep the previous attempt as a historical row with
flag = NO and insert the new attempt with flag = YES. When both attempts carry the same
end_time, the historical row also matches max(end_time) and the query returns two rows
for one task code. On MySQL t_ds_task_instance.end_time is a datetime without
fractional seconds, so a task which fails and is retried quickly can easily produce this.

The only caller, DependentExecute#dependResultByAllTaskOfWorkflowInstance, keys the
result by task code with Collectors.toMap and no merge function, so the duplicate throws
IllegalStateException: Duplicate key and the dependency evaluation of a DEPENDENT task
configured with "ALL tasks" fails.

Brief change log

• TaskInstanceMapper.xml#findLastTaskInstances: only consider the valid instance of each
  task code, by applying flag = 1 both to the grouped rows and to the outer matched row.
  The ordering is unchanged: the last instance is still the one with the greatest end_time.
• Added a TaskInstanceDaoImplTest case covering a historical attempt and its valid retry
  sharing an end_time.

Verify this pull request

This change added tests and can be verified as follows:

• Added
  TaskInstanceDaoImplTest#queryLastTaskInstanceListIntervalInWorkflowInstanceWhenAttemptsShareTheSameEndTime,
  which inserts a FAILURE attempt with flag = NO and its SUCCESS retry with
  flag = YES, both with the same end_time, and asserts that only the valid attempt is
  returned.

./mvnw -pl dolphinscheduler-dao -am clean test \
    -Dtest=TaskInstanceDaoImplTest \
    -Dsurefire.failIfNoSpecifiedTests=false

Verified locally:

• The new test fails without the flag = 1 conditions with expected: <1> but was: <2>
  and passes with this change.
• The two pre-existing TaskInstanceDaoImplTest cases still pass.
• The full dolphinscheduler-dao suite passes: 284 tests, 0 failures, 0 errors.
• ./mvnw -pl dolphinscheduler-dao spotless:check passes.

Pull Request Notice

Pull Request Notice

If your pull request contains incompatible change, you should also add it to docs/docs/en/guide/upgrade/incompatible.md

🤖 Generated with Claude Code

[SbloodyS]

In the current execution logic, there is only one running task instance under the same task code in the same workflow instance. We should find out why there is dirty data, rather than avoid this problem.

[SEPURI-SAI-KRISHNA]

Thanks for the review. I agree there should only ever be one running task instance per task code, but the rows this query returns are not running instances, and the duplicates are not dirty data.

The query only looks at finished attempts. findLastTaskInstances filters state != 8 (NEED_FAULT_TOLERANCE) and matches on end_time, so a running instance (end_time is null) can never be selected. Its only caller reaches it after the depended-on workflow has already finished:

// DependentExecute#calculateResultForTasks
if (workflowInstance.getState().isSuccess()) {
    ...
    taskInstanceDao.queryLastTaskInstanceListIntervalInWorkflowInstance(workflowInstance.getId(), ...)

So there is no running instance involved in this path at all.

Multiple finished rows per task code are created by the engine on purpose. Retry, failover and failed-recover each insert a brand new t_ds_task_instance row with the same task_code / workflow_instance_id and flip the previous row to flag = NO:

// RetryTaskInstanceFactory#createTaskInstance
final TaskInstance taskInstance = cloneTaskInstance(needRetryTaskInstance);
taskInstance.setId(null);
...
taskInstanceDao.insert(taskInstance);
needRetryTaskInstance.setFlag(Flag.NO);
taskInstanceDao.updateById(needRetryTaskInstance);

FailoverTaskInstanceFactory and FailedRecoverTaskInstanceFactory do the same. That is exactly why this query exists and is called findLast… — it is supposed to pick the newest of several legitimate attempts.

The defect is the tie-break, not the data. The query groups by task_code taking max(end_time), then joins back on equality:

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

end_time is not unique. On MySQL t_ds_task_instance.end_time is a plain datetime, second precision, no fractional part:

`end_time` datetime DEFAULT NULL COMMENT 'task end time',

So a short task that fails and is retried (or failed over) inside the same second produces two attempts with byte-identical end_time, both of them equal to the max, and the join emits both. DependentExecute then does:

.collect(Collectors.toMap(TaskInstance::getTaskCode, TaskInstance::getState));

Collectors.toMap has no merge function, so the dependent task fails the whole workflow with IllegalStateException: Duplicate key. It is a genuine plan-level ambiguity, not corrupted rows, the same two rows are perfectly valid to every other query.

The singular sibling in the same mapper already handles this correctly and can't return duplicates:

<select id="findLastTaskInstance" ...>
  ... order by end_time desc limit 1

The plural version is simply inconsistent with it.

Why max(id). Attempts are only ever created from an already-terminated attempt (failover marks the old row NEED_FAULT_TOLERANCE, which this query excludes), so the auto-increment id is a strictly monotonic attempt order and max(id) is deterministic where max(end_time) is not. It selects the same row as before whenever end_time values differ, it only makes the tied case well-defined.

If you'd rather keep end_time as the primary ordering and use the id only as a tie-break, I'm happy to change it to max(id) restricted to the rows holding max(end_time); the fix is one query either way. And if you still think the duplicate rows themselves are the bug, I'd like to understand which factory you consider wrong to insert a second finished row — from the code above it looks intentional.

The added test reproduces it: it fails on dev with expected: <1> but was: <2> and passes with this change.

[SbloodyS]

In the current execution logic, there is only one running task instance under the same task code in the same workflow instance. We should find out why there is dirty data, rather than avoid this problem.

As I said before, this is not the root cause. If there is no dirty data, this problem will not appear.

[ruanwenjun]

Do all instances of the same task have end times within the same second?

[SEPURI-SAI-KRISHNA]

@ruanwenjun

Do all instances of the same task have end times within the same second?

No, and they don't need to. Two attempts of one task code ending in the same second is enough to
break the whole dependency check, because DependentExecute collects every task of the workflow into
one map:

.collect(Collectors.toMap(TaskInstance::getTaskCode, TaskInstance::getState));

A single duplicated key throws, so one tied pair among a hundred well-spaced tasks fails the
dependent task for the entire upstream workflow.

When two attempts do tie:

• failRetryInterval is allowed to be 0 (use-failed.ts, props: { min: 0 }), so an immediate
  retry is a supported configuration, and a task that fails fast (bad command, connection refused,
  a failure raised before the task really starts) can fail and be retried inside the same second.

• On MySQL the tie is much easier than "same millisecond", because
  t_ds_task_instance.end_time is a datetime with no fractional seconds:

  `end_time` datetime DEFAULT NULL COMMENT 'task end time',

  Any sub-second difference between the two attempts is erased on write, so they only have to land
  in the same wall-clock second.

To be straight about the exposure: on PostgreSQL end_time is a timestamp with microsecond
precision, so a tie there is far less likely. This is mostly a MySQL problem, and it needs a fast
retry. It is narrow, but it is reachable with a stock configuration, and the failure mode is a hard
exception rather than a wrong result.

[SEPURI-SAI-KRISHNA]

@SbloodyS

As I said before, this is not the root cause. If there is no dirty data, this problem will not appear.

I think we're disagreeing about which data counts as dirty, so I made the reproduction match the
engine exactly rather than keep arguing. The test now writes the two rows the way
RetryTaskInstanceFactory writes them: the superseded attempt gets flag = NO, the new attempt gets
flag = YES.

// 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);

There is now exactly one valid task instance for that task code, which is the invariant you're
describing. On dev the query still returns both rows:

TaskInstanceDaoImplTest.queryLastTaskInstanceListIntervalInWorkflowInstanceWhenAttemptsShareTheSameEndTime:86
expected: <1> but was: <2>

No row in that fixture is dirty. Both are written by the engine on purpose, the superseded one is
correctly invalidated, and the query returns both anyway because it never looks at flag and because
end_time is not unique. The old attempt also can't just be deleted, the retry history is what the
task instance list shows.

If your point is that the query should be honouring the flag = YES invariant instead of picking the
newest row, I'm glad to do it that way. Both of these turn the test green:

-- (a) current PR: deterministic newest attempt
select max(id) as max_id from t_ds_task_instance ... group by task_code

-- (b) rely on the engine invariant instead
... and flag = 1

I chose (a) because it holds even if the invariant is ever violated, and because the singular sibling
in the same mapper, findLastTaskInstance, already breaks the tie the same way with
order by end_time desc limit 1. Happy to switch to (b), or apply both, whichever you prefer.

And if you do think a second finished row for one task code shouldn't exist at all, could you point
me at the factory you consider wrong? RetryTaskInstanceFactory, FailoverTaskInstanceFactory and
FailedRecoverTaskInstanceFactory all deliberately insert a new row instead of updating the old
one, so from the code this looks like the intended design. If it isn't, that's a much larger bug than
this one and I'd rather raise it separately than leave DependentExecute throwing in the meantime.

[SbloodyS]

flag = YES stands for a valid instance, and NO stands for a historical instance used for recovery failure and other processes. You provided an example of dirty data. Please provide the steps of how to reproduce this scenario. @SEPURI-SAI-KRISHNA

[SEPURI-SAI-KRISHNA]

@SbloodyS Thanks, that definition helps, and I've now made the query follow it (below).

On the fixture first: it contains one flag = YES row and one flag = NO row, i.e. exactly one
valid instance plus one historical instance, which is the state RetryTaskInstanceFactory leaves
behind. By your own definition that's clean data. The only unusual thing about it is that the two
rows carry the same end_time, and that part isn't written by the engine, it's what the column does
to two different timestamps: on MySQL t_ds_task_instance.end_time is a datetime with no
fractional seconds, and MySQL rounds a fractional value to whole seconds on insert (5.6.4+). Two
attempts a few hundred ms apart can therefore be stored with an identical end_time.

Steps to reproduce

Metadata DB must be MySQL (see the note at the end for why).

1. Workflow A, one SHELL task t1 whose first attempt fails and whose retry succeeds, so that
   A finishes SUCCESS (required: DependentExecute only evaluates an upstream instance that
   succeeded):

   test -f /tmp/ds18543.flag && exit 0
   touch /tmp/ds18543.flag
   exit 1

   Set Number of failed retries = 1 and Failed retry interval = 0, so the retry starts
   immediately and both attempts land in the same second.

2. Run A. It ends SUCCESS with two task instances for t1.

3. Check what was stored:

   select id, task_code, state, flag, end_time
   from t_ds_task_instance
   where workflow_instance_id = <A_instance_id>
   order by id;

   The failed attempt is flag = 0, the retry is flag = 1, and both rows show the same
   end_time whenever the two attempts round into the same stored second.

4. Workflow B with a DEPENDENT task on workflow A with ALL tasks selected
   (DEPENDENT_ALL_TASK_CODE). Run B.

5. The dependent check fails in the master with

   java.lang.IllegalStateException: Duplicate key <taskCode> (attempted merging values ... and ...)
   

   instead of resolving the dependency.

Being straight about it: step 3 is timing dependent. If the two attempts happen to straddle a second
boundary you get two distinct end_time values and have to run it again (rm /tmp/ds18543.flag
between runs). That is exactly why the unit test constructs the stored state directly instead of
racing the clock. What is not timing dependent is the DAO behaviour once two such rows exist, and
that is what the test pins down.

I also want to be upfront that I found this by reading the query rather than from a production
incident, and that it is mainly a MySQL exposure: on PostgreSQL end_time is a timestamp with
microsecond precision, so the collision is far less likely there.

What I changed

Since flag = YES is the valid instance, the query should say so, so I added it:

select max(id) as max_id
from t_ds_task_instance
where 1=1
and workflow_instance_id = #{workflowInstanceId}
and state != 8
and flag = 1
and end_time is not null
...
group by task_code

flag = 1 enforces the invariant you described, and max(id) keeps the result deterministic if two
valid rows ever do coexist. Full dolphinscheduler-dao suite passes, 284 tests.

If you would rather have only and flag = 1 and keep max(end_time) as the join key, I'm happy to
push that instead, just say the word.

[SEPURI-SAI-KRISHNA]

Thanks @xiangzihao for merging dev into the branch. That triggered the first full CI run on this PR, and the result is green where it matters:

• Unit-Test (dolphinscheduler-dao | Java 8) — success
• Unit-Test (dolphinscheduler-dao | Java 11) — success

dolphinscheduler-dao is the only module this PR touches, and both runs include the new
TaskInstanceDaoImplTest case. Build, CodeQL, E2E, all four cluster-tests and all six schema-checks also passed.

One job is red, but it never reached a test:

Unit-Test (dolphinscheduler-spi | Java 8)
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-surefire-plugin:3.5.4:test
  Could not transfer artifact org.apache.maven.surefire:surefire-junit-platform:pom:3.5.4
  from/to central (https://repo.maven.apache.org/maven2):
  Remote host terminated the handshake: SSL peer shut down incorrectly

That is Maven Central dropping the TLS handshake while resolving a surefire plugin POM, in a module this PR does not touch. Because the matrix is fail-fast it then cancelled ten sibling jobs. Could someone re-run the failed jobs when convenient? I don't have permission to do it myself.

@SbloodyS, whenever you have a moment: the query now follows the definition you gave. It filters flag = 1 so only the valid instance of each task code is considered, and it joins on max(id) rather than on end_time, so the grouped subquery returns exactly one row per task code by construction and a tie on end_time can no longer produce a duplicate key. Happy to adjust further if you would still prefer a different approach.

[SbloodyS]

Adding flag = 1 fixes the reported path: retry, failover, and recovery intentionally retain the previous attempt as flag = NO, and the old query incorrectly allowed that historical row to match the same max(end_time) as the valid attempt.

However, max(id) adds a separate fallback that silently selects one row if multiple flag = YES instances exist. According to the valid-instance invariant discussed above, that state is invalid and should remain observable instead of being hidden. It also changes the existing ordering semantics from the greatest end_time to the greatest insertion ID.

Please scope the fix to selecting only the valid, finished instance, or retain the existing end_time ordering while applying flag = 1 to both the grouped rows and the outer matched row. If ordering by ID is intentional, please add a test where ID order and end_time order differ and document the semantic change.

[SEPURI-SAI-KRISHNA]

Thanks @SbloodyS. I went with the second option and removed max(id).

The query is now the one on dev with flag = 1 added in two places:

• in the grouped subquery, so max(end_time) is taken over valid instances only
• on the outer matched row, so a historical flag = NO attempt that shares that end_time no longer joins

The join is back on workflow_instance_id, task_code and end_time, so "last" means the greatest end_time, as before. Among flag = YES rows the query now behaves exactly as it does on dev: if two of them ever share an end_time, both are returned instead of one being picked. I also removed the end_time is not null filter I had added. The equality join already excludes unfinished instances, so it was redundant.

The regression test is unchanged: a FAILURE attempt with flag = NO and its SUCCESS retry with flag = YES, both with the same end_time. With the two flag = 1 conditions removed it fails with expected: <1> but was: <2>, and with them it passes. The full dolphinscheduler-dao suite passes locally (284 tests), and I've updated the PR description to match.

[SbloodyS]

LGTM

[ruanwenjun]

@ruanwenjun

Do all instances of the same task have end times within the same second?

No, and they don't need to. Two attempts of one task code ending in the same second is enough to break the whole dependency check, because DependentExecute collects every task of the workflow into one map:

.collect(Collectors.toMap(TaskInstance::getTaskCode, TaskInstance::getState));

A single duplicated key throws, so one tied pair among a hundred well-spaced tasks fails the dependent task for the entire upstream workflow.

When two attempts do tie:

• failRetryInterval is allowed to be 0 (use-failed.ts, props: { min: 0 }), so an immediate
  retry is a supported configuration, and a task that fails fast (bad command, connection refused,
  a failure raised before the task really starts) can fail and be retried inside the same second.

• On MySQL the tie is much easier than "same millisecond", because
  t_ds_task_instance.end_time is a datetime with no fractional seconds:

  `end_time` datetime DEFAULT NULL COMMENT 'task end time',

  Any sub-second difference between the two attempts is erased on write, so they only have to land
  in the same wall-clock second.

To be straight about the exposure: on PostgreSQL end_time is a timestamp with microsecond precision, so a tie there is far less likely. This is mostly a MySQL problem, and it needs a fast retry. It is narrow, but it is reachable with a stock configuration, and the failure mode is a hard exception rather than a wrong result.

I still cannot understand, why two task instances belong to the same task code can have the same end time at one workflow instance, could this really happen?

[ruanwenjun]

LGTM

[SEPURI-SAI-KRISHNA]

@ruanwenjun Thanks for the review. Yes, it can happen, although it is narrow. It needs two conditions together:

1. A retry right after a quick failure. The failed attempt's row is kept with flag = NO, and the retry is inserted as a new
   row with the same task_code and workflow_instance_id. With Failed retry interval = 0, which the
   UI allows, a task that fails immediately (bad command, connection refused) has its retry start at once.
2. A MySQL metadata database. t_ds_task_instance.end_time is a datetime with no fractional
   seconds, so MySQL rounds each end time to a whole second. If both attempts' end times round to the
   same second, the stored values are identical.

The step-by-step reproduction is in this comment. It is timing dependent, which is why the test builds the stored rows directly.

The fix does not depend on how often this happens, though. The flag = NO row is a historical attempt and should never be the "last task instance", so filtering flag = 1 is correct even when the end times differ.