The remote task tracked by the trigger failed due to the reassignment of the trigger task

[TakawaAkirayo]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.7.2

What happened?

The trigger instance handled by triggerer-2 was canceled due to reassignment to triggerer-1.
As a result, triggerer-2 cleaned up the local trigger instance, which terminated the remote job using its ID.
The trigger instance in triggerer-2 keep polling the status of remote job with the same id, but detected that the remote job was killed, so completed its run, and exited.
Consequently, the remote job failed due to the reassignment.

Production log summarized by time sequence

Time: 2024-07-11, 00:00:12 -07
Event: Start to run on worker
Runtime: worker
Log: {task_command.py:416} INFO - Running <TaskInstance: f67fb155-6ff2-4fde-bbb2-6f0ef16af05e.91fade50-a3da-4966-a0ff-44c24d59303a scheduled__2024-07-11T06:00:00+00:00 [running]> on host worker-host-3
----------------------------------

Time: 2024-07-11, 00:00:14 -07
Event: Defer to trigger
Runtime: worker
Log: Pausing task as DEFERRED. dag_id=f67fb155-6ff2-4fde-bbb2-6f0ef16af05e, task_id=91fade50-a3da-4966-a0ff-44c24d59303a, execution_date=20240711T060000, start_date=20240711T070011
----------------------------------

Time: 2024-07-11T00:00:54.788-0700
Event: Trigger instance-1 started
Runtime: triggerer-2
Log: {triggerer_job_runner.py:595} INFO - trigger f67fb155-6ff2-4fde-bbb2-6f0ef16af05e/scheduled__2024-07-11T06:00:00+00:00/91fade50-a3da-4966-a0ff-44c24d59303a/-1/1 (ID 78297) starting
----------------------------------

Time: 2024-07-11T00:00:54.788-0700
Event: Trigger instance-1 check job status
Runtime: triggerer-2
Log: {datax_trigger.py:26} INFO - Check if job finished.
----------------------------------

Time: 2024-07-11T00:00:54.823-0700
Event: Trigger instance-1 get job running
Runtime: triggerer-2
Log: {datax_trigger.py:32} INFO - {"code":"0000","message":"success","result":{"dagRunId":"manual__2024-07-11T07:00:43.473585+00:00","startTime":"2024-07-11T00:00:14+00:00","endTime":null,"status":"RUNNING","extraInfo":{"dagId":"f67fb155-6ff2-4fde-bbb2-6f0ef16af05e-61566","status":"<strong style=\"color:blue;\">RUNNING</strong>"}}}
----------------------------------

Time: 2024-07-11T00:01:00.503-0700
Event: Triggerer-2 cancel trigger due to find the trigger has been occupied by another triggerer process
Runetime: triggerer-2
Log: {triggerer_job_runner.py:607} ERROR - Trigger cancelled; message=
----------------------------------

Time: 2024-07-11, 00:01:00 -07
Event: Kyuubi Trigger cleanup called
Runtime: triggerer-2
Log: {datax_trigger.py:51} INFO - Trigger is cancelled, clean up now.
----------------------------------

Time: 2024-07-11T00:01:00.977-0700
Event: Trigger instance-1 cleanup
Runeimte: triggerer-2
Log: {triggerer_job_runner.py:621} INFO - trigger f67fb155-6ff2-4fde-bbb2-6f0ef16af05e/scheduled__2024-07-11T06:00:00+00:00/91fade50-a3da-4966-a0ff-44c24d59303a/-1/1 (ID 78297) completed
----------------------------------

Time: 2024-07-11, 00:02:00 -07
Event: Trigger check if job success
Runtime: triggerer-1
Log: {datax_trigger.py:26} INFO - Check if job finished.
----------------------------------

Time: 2024-07-11, 00:02:00 -07
Event: check job status
Runtime: triggerer-1
Log: {datax_trigger.py:32} INFO - {"code":"0000","message":"success","result":{"dagRunId":"manual__2024-07-11T07:00:43.473585+00:00","startTime":"2024-07-11T00:00:14+00:00","endTime":null,"status":"FAILED","extraInfo":{"executionLogAddress":"<a target=\"_blank\" href=\"https://host/api/v1/logs?JobId=79028\">Execution Log</a>","dagId":"f67fb155-6ff2-4fde-bbb2-6f0ef16af05e-61566","status":"<strong style=\"color:red;\">FAILED</strong>"}}}
----------------------------------

What you think should happen instead?

Triggers will be canceled and cleaned up by the current triggerer process if the current triggerer finds that those running trigger instances have been reassigned to other triggerers.
However, if the cleanup is called prematurely while the trigger is still in an active state, it could impact the same trigger when reassigned to another triggerer process.
Therefore, introducing the capability to disable this cleanup behavior in such scenarios.

How to reproduce

1. Set the triggerer_health_check_threshold to a small number like 3(s).
2. Make triggerer instances at least 2.
3. Use blocking io like http request in the trigger's run method.
4. Trigger multiple deferred tasks(for our cases: 100+ trigger jobs) in the same time.

Operating System

Ubuntu

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[tirkarthi]

Can you please add sample code to reproduce this along with the trigger implementation? Does the trigger implement cleanup method?

[TakawaAkirayo]

@tirkarthi Below code can reproduce this issue along with the trigger's implementation with clean up
I raised a PR for this issue: https://github.com/apache/airflow/pull/41232/files#diff-02359b2cc957e3a9be05ce055322558b36359c13572afc38f14f62a6dfae01c3

class RemoteJob:
    RUNNING_STATUS = "running"
    FAILED_STATUS = "failed"
    SUCCESS_STATUS = "success"

    def __init__(self, status):
        self.status = status

    def get_status(self):
        return self.status

    def set_status(self, status):
        self.status = status

    def kill(self):
        self.set_status(RemoteJob.FAILED_STATUS)


class RemoteService:
    _instance = None

    event_queue: list[tuple[RemoteJobTrigger, int]]
    remote_jobs: dict[int, RemoteJob]

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.event_queue = []
            cls._instance.remote_jobs = {
                1: RemoteJob(RemoteJob.RUNNING_STATUS),
            }
        return cls._instance

    def complete_job(self, job_id: int):
        self.remote_jobs[job_id].set_status(RemoteJob.SUCCESS_STATUS)

    def kill_job(self, caller: RemoteJobTrigger, job_id: int):
        self.event_queue.append((caller, job_id))
        self.remote_jobs[job_id].kill()

    def get_job(self, job_id: int):
        return self.remote_jobs[job_id]

    def reset(self):
        self.event_queue = []
        self.remote_jobs = {
            1: RemoteJob(RemoteJob.RUNNING_STATUS),
        }


class TriggerInstances:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.trigger_instances = []
        return cls._instance

    def reset(self):
        self.trigger_instances.clear()


class RemoteJobTrigger(BaseTrigger):
    def __init__(self, remote_job_id):
        super().__init__()
        self.remote_job_id = remote_job_id
        self.finished = False
        self.cleanup_done = False
        self.last_status = None
        TriggerInstances().trigger_instances.append(self)

    def get_status(self):
        return RemoteService().get_job(self.remote_job_id).get_status()

    async def run(self):
        print(f"Trigger object {id(self)}: start to run")

        while self.get_status() == RemoteJob.RUNNING_STATUS:
            await asyncio.sleep(0.1)
        self.finished = True
        print(f"Trigger object {id(self)}: finished with status: {self.get_status()}")
        yield TriggerEvent(self.remote_job_id)

    async def cleanup(self) -> None:
        RemoteService().kill_job(self, self.remote_job_id)
        self.cleanup_done = True
        print(f"Trigger object {id(self)}: cleanup done")

    def has_been_cleaned(self):
        return self.cleanup_done

    def failed(self):
        return self.get_status() == RemoteJob.FAILED_STATUS

    def serialize(self):
        return (
            "tests.jobs.test_triggerer_job.RemoteJobTrigger",
            {"remote_job_id": self.remote_job_id},
        )


@pytest.mark.asyncio
async def test_disable_trigger_cleanup_on_reassigned_to_other_triggerers(session, tmp_path):

    from typing import Callable, Awaitable
    from airflow.configuration import conf
    from airflow.jobs.triggerer_job_runner import TriggerDetails

    remote_service = RemoteService()
    # The trigger will be created sequentially,
    # so we can use a global variable to track the trigger instances
    trigger_instances_holder = TriggerInstances()

    # Build the scenario that the trigger reassigned to another triggerer
    # due to the first triggerer missed heartbeat
    async def make_triggerer2_occupy_the_trigger_of_triggerer1(
        check: Callable[[TriggerDetails, TriggerDetails], Awaitable[None]]
    ):

        # Create trigger to check the remote job status
        trigger = RemoteJobTrigger(1)
        # Remove this one from the trigger_instances_holder, we only need to track the instances
        # that created by the triggerer job runner
        del trigger_instances_holder.trigger_instances[0]

        trigger_orm = Trigger.from_object(trigger)
        trigger_orm.id = 1
        session.add(trigger_orm)

        dag = DagModel(dag_id="test-dag")
        dag_run = DagRun(dag.dag_id, run_id="abc", run_type="none")
        ti = TaskInstance(
            PythonOperator(task_id="dummy-task", python_callable=print),
            run_id=dag_run.run_id,
            state=TaskInstanceState.DEFERRED,
        )
        ti.dag_id = dag.dag_id
        ti.trigger_id = 1
        session.add(dag)
        session.add(dag_run)
        session.add(ti)

        job1 = Job()
        health_check_threshold_value = conf.getfloat("triggerer", "triggerer_health_check_threshold")
        # triggerer1 miss heartbeat due to high host load, so triggerer2 can pick up the trigger
        job1.latest_heartbeat = timezone.utcnow() - datetime.timedelta(
            seconds=health_check_threshold_value + 1)

        job2 = Job()
        session.add(job1)
        session.add(job2)

        session.commit()

        job_runner1 = TriggererJobRunner(job1)
        job_runner2 = TriggererJobRunner(job2)

        # job_runner1 load triggers, it will pick up the trigger
        job_runner1.load_triggers()
        assert len(job_runner1.trigger_runner.to_create) == 1
        # Create the trigger and start running
        await job_runner1.trigger_runner.create_triggers()
        assert len(job_runner1.trigger_runner.triggers) == 1

        job2.heartbeat(lambda x: {}, session)
        job_runner2.load_triggers()
        assert len(job_runner2.trigger_runner.to_create) == 1
        await job_runner2.trigger_runner.create_triggers()
        # Now job_runner2 will take over the trigger for execution
        assert len(job_runner2.trigger_runner.triggers) == 1

        # Globally, there are two trigger instances
        assert len(trigger_instances_holder.trigger_instances) == 2

        # job_runner2 performs heartbeat normally
        job2.heartbeat(lambda x: {}, session)

        # Let the job_runner2 start to run
        job_runner2.trigger_runner.daemon = True
        job_runner2.trigger_runner.start()

        # Now the job_runner1 will find that the trigger has been reassigned to someone else
        # and will cancel the trigger
        job_runner1.load_triggers()
        job_runner1.trigger_runner.daemon = True
        job_runner1.trigger_runner.start()

        try:
            trigger_details_of_job1 = job_runner1.trigger_runner.triggers[1]
            trigger_details_of_job2 = job_runner2.trigger_runner.triggers[1]
            await check(trigger_details_of_job1, trigger_details_of_job2)
        finally:
            job_runner1.trigger_runner.stop = True
            job_runner1.trigger_runner.join(30)

            job_runner2.trigger_runner.stop = True
            job_runner2.trigger_runner.join(30)

            # clean up state
            remote_service.reset()
            trigger_instances_holder.reset()
            session.delete(trigger_orm)
            session.delete(ti)
            session.delete(dag_run)
            session.delete(dag)
            session.commit()

    async def expect_the_trigger_will_be_cleanup_on_reassigned(
        trigger_details1: TriggerDetails, trigger_details2: TriggerDetails
    ):
        # The trigger of job_runner2 should finish because the trigger of job_runner1 has been cancelled,
        # and it will terminate the remote job when cleanup is called
        trigger_of_job_2 = trigger_instances_holder.trigger_instances[1]
        while trigger_of_job_2.has_been_cleaned() is False:
            await asyncio.sleep(0.1)
        # If reach here, the trigger of job_runner2 should be finished and cleaned up
        assert trigger_of_job_2.finished is True
        # The remote job should fail due to the trigger of job_runner1 has been cancelled,
        # and it has cleaned up the same remote job
        assert trigger_of_job_2.failed() is True

        # The cancelled trigger should already be cleaned up
        cancelled_trigger = trigger_instances_holder.trigger_instances[0]
        assert cancelled_trigger.has_been_cleaned() is True

        # The remote job should have failed because the trigger of job_runner1 was cancelled
        assert remote_service.get_job(1).get_status() == RemoteJob.FAILED_STATUS

        # There should be two kill events, each from a different trigger instance
        # managed by separate triggerers
        assert len(remote_service.event_queue) == 2
        # The first kill job request comes from the canceled trigger of job_runner1
        (cancel_caller, remote_job_id) = remote_service.event_queue[0]
        assert cancel_caller == cancelled_trigger and remote_job_id == 1

    await make_triggerer2_occupy_the_trigger_of_triggerer1(expect_the_trigger_will_be_cleanup_on_reassigned)

[tirkarthi]

cleanup method is called when trigger is cancelled due to triggerer restart, reassignment of trigger etc. You need to implement something like safe_to_cancel method in the linked PR before doing the actual cancellation.

#39442

[TakawaAkirayo]

cleanup method is called when trigger is cancelled due to triggerer restart, reassignment of trigger etc. You need to implement something like safe_to_cancel method in the linked PR before doing the actual cancellation.

#39442

Many thanks for the suggestion, safe_to_cancel is a good idea, I think give the trigger more context like why this is canceled could help

[EMT-Swaroop]

Hey Team,

We are using Airflow 2.9.3 version in dev, preprod and prod. But in preprod and prod triggerer is continuously getting restarted but not in dev. We have checked the resource limits, and the consumption is normal for CPU and memory. No DAG changes have also been done. Nothing from k8s side as well.

Below are the logs we are getting

[Info] Handling Signal: term
[Info] Worker exiting (pid: 24)
[Info] Worker exiting (pid: 25)
{triggerer_job_runner.py: 350} Exited Trigger Loop

We are using the default probe values from parent chart.

Please let me know if there is any reason or fix.

[github-actions]

This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.