[Fix-18543][DAO] Return only the last 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 joining on
instance.end_time = t_max.max_end_time. end_time is not unique, so when two attempts
of the same task share an end_time the query returns two rows for one task code.

On MySQL t_ds_task_instance.end_time is a datetime without fractional seconds, so
every value is truncated to a whole second and a task which fails and is retried quickly
ends up with both attempts carrying an identical end_time. The query filters on
state != 8 only, not on flag, so the invalidated previous attempt is included too.

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: resolve the last attempt per task code
  with max(id) and join on the primary key, which yields exactly one row per task code.
• Kept the previous behaviour of ignoring attempts which have not finished, by filtering
  end_time is not null in the sub-query. Previously this was implicit — max(end_time)
  ignores NULLs and instance.end_time = t_max.max_end_time never matches a NULL.
• Added a TaskInstanceDaoImplTest case covering two attempts sharing an end_time.

A note on max(id) vs max(end_time)

t_ds_task_instance.id is an auto-increment primary key, so the highest id for a
(workflow_instance_id, task_code) pair is the most recently created attempt. A retry
row is only inserted after the previous attempt has ended, so for retries the latest id
and the latest end_time identify the same row — but only id is unique, so only id
can guarantee a single row. This also removes the non-aggregated workflow_instance_id
from the group by task_code sub-query, which is friendlier to ONLY_FULL_GROUP_BY on
MySQL.

Happy to switch to a different tie-break if maintainers prefer to keep end_time as the
ordering key.

Verify this pull request

This change added tests and can be verified as follows:

• Added
  TaskInstanceDaoImplTest#queryLastTaskInstanceListIntervalInWorkflowInstanceWhenAttemptsShareTheSameEndTime,
  which inserts a FAILURE attempt and its SUCCESS retry with the same end_time and
  asserts a single row — the last attempt — is returned.

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

Verified locally:

• The new test fails on dev with expected: <1> but was: <2> and passes with this
  change.
• The two pre-existing TaskInstanceDaoImplTest cases still pass, so the "last attempt"
  semantics are unchanged for the normal, non-tied case.
• 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.