Return 410 for stale id in set rtif API

[FrankYang0529]

Why

A task whose operator sets overwrite_rtif_after_execution=True logs a 404 error when setting rtif. The server regenerates the task instance id (archiving the old one to history) via a retry or a clear. However, the worker's finalize() still overwrites
RTIF using the stale id.

How

When a TI id is missing from the live table but present in history, PUT /task-instances/{id}/rtif now returns 410 Gone instead of 404. The hearbeat API also uses this mechanism. The supervisor treats that 410 as an expected, skippable outcome (debug log, no error). A genuine 404 still propagates as a real error.

closes: #66416

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code with Opus 4.8

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[SameerMesiah97]

Looks good. Just one nit.

[jason810496]

Nice! LGTM overall, thanks.

[jason810496]

Hi @FrankYang0529,
Would you mind to test the minimal reproduction mentioned in #66416 when you have a moment? Thanks.

[FrankYang0529]

@jason810496 Thanks for the review. I tested with following dag and we don't have error message in the first try Post Execute now.

from __future__ importannotationsimportdatetimefromairflow.sdkimportDAGfromairflow.sdk.bases.operatorimportBaseOperatorclassFailingOverwriteRTIFOperator(BaseOperator):
template_fields= ("message",)
overwrite_rtif_after_execution=Truedef__init__(self, *, message: str="hello {{ ds }}", **kwargs):
super().__init__(**kwargs)
self.message=messagedefexecute(self, context):
raiseRuntimeError("Intentional failure to reproduce RTIF finalize bug")
withDAG(
dag_id="repro_rtif_finalize",
start_date=datetime.datetime(2024, 1, 1),
schedule=None,
catchup=False,
):
FailingOverwriteRTIFOperator(
task_id="boom",
retries=1,
retry_delay=datetime.timedelta(seconds=5),
)

[jason810496]

Nice!

def__init__(self, *, message: str="hello {{ ds }}", **kwargs):

Hope you haven't terminated the cluster yet. Could we test with def __init__(self, *, message: str = "hello {{ ti.try_number }}", **kwargs): when you have a moment to double check are we able to get the Intentional failure to reproduce RTIF finalize bug sentinel error or not, thanks.

[FrankYang0529]

Hope you haven't terminated the cluster yet. Could we test with def __init__(self, *, message: str = "hello {{ ti.try_number }}", **kwargs): when you have a moment to double check are we able to get the Intentional failure to reproduce RTIF finalize bug sentinel error or not, thanks.

Thanks for the great suggestion. Yes, we can get expected message.

[jason810496]

Thanks for the great suggestion. Yes, we can get expected message.

Thanks!

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test