You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "TakawaAkirayo (via GitHub)" <gi...@apache.org> on 2024/08/03 06:43:52 UTC
[PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
TakawaAkirayo opened a new pull request, #41232:
URL: https://github.com/apache/airflow/pull/41232
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
### 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.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716589260
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +647,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details.get("termination_reason")):
Review Comment:
Sure, already applied.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716800188
##########
airflow/triggers/base.py:
##########
@@ -115,6 +116,19 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ def should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ """
+ Check the trigger should be cleaned up or not base on the context.
+
+ :param termination_reason: The reason for terminating the trigger.
+ Since the trigger could be terminated for various reasons, like triggerer restart or reassigned to
+ another triggerer, this method allows the trigger to decide whether it should be cleaned up under the
+ current circumstances.
+
+ By default, will always return True, override this method base custom requirements.
Review Comment:
Add one section in the Deferrable Operators & Triggers page, please help to review @uranusjr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717895551
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +756,31 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
Review Comment:
Thanks, renamed to _add_trigger_cancel_reasons already
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717761389
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +648,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details["termination_reason"]):
Review Comment:
One reason I added the new method is that I didn’t want to modify the existing cleanup interface, which allows for backward compatibility with user code (please correct me if I’m wrong). The downside is that it’s not as simple and direct as you mentioned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717799345
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +648,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details["termination_reason"]):
Review Comment:
And that is a very reasonable concern @TakawaAkirayo. However right now with airflow 3 coming, we have a special opportunity where we can break backward compatibility. Even in 2.x we could have done it by adding some kind of compatibility layer / conditional handling. But since this will only make it into 3.x we don't need that. What we will need to do though, is update providers to be forward-compatible with this change, so that new provider releases will be compat with 2.x and 3.x as much as possible. E.g. maybe changing existing trigger cleanup signatures to take **kwargs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717888649
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +756,31 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
Review Comment:
Oh! should remove it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716328140
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -538,10 +539,13 @@ async def cancel_triggers(self):
This allows the cleanup job to delete them.
"""
while self.to_cancel:
- trigger_id = self.to_cancel.popleft()
+ trigger_id, reason = self.to_cancel.popleft()
if trigger_id in self.triggers:
+ # Add more context here before cancelling
+ to_be_cancelled = self.triggers[trigger_id]
+ to_be_cancelled["termination_reason"] = reason
# We only delete if it did not exit already
- self.triggers[trigger_id]["task"].cancel()
+ to_be_cancelled["task"].cancel()
Review Comment:
I couldn’t find where this `to_be_cancelled` is used. Why do we store things into this dict that’s never read by anyone?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716332040
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +757,28 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
+ # find out reassigned triggers
+ reassigned_trigger_ids = set(Trigger.filter_out_reassigned_triggers(triggerer_id, cancel_trigger_ids))
Review Comment:
Why does `filter_out_reassigned_triggers` return a list and this immediately converts it into a set? Why not just return a set directly instead?
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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:
Review Comment:
These singleton classes (this one and the one below) are a bit overkill IMO. They can just be regular classes and instantiate regular test-scoped objects. It’d be much easier to debug too.
##########
airflow/triggers/base.py:
##########
@@ -115,6 +116,19 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ def should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ """
+ Check the trigger should be cleaned up or not base on the context.
+
+ :param termination_reason: The reason for terminating the trigger.
+ Since the trigger could be terminated for various reasons, like triggerer restart or reassigned to
+ another triggerer, this method allows the trigger to decide whether it should be cleaned up under the
+ current circumstances.
+
+ By default, will always return True, override this method base custom requirements.
Review Comment:
Is this only for testing purposes? The docstring should say so instead if that’s the case.
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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()
+
+ def add_instances(self, instance):
+ self.trigger_instances.append(instance)
+
+
+class RemoteJobTrigger(BaseTrigger):
+ def __init__(self, remote_job_id, cleanup_on_reassignment: bool):
+ super().__init__()
+ self.remote_job_id = remote_job_id
+ self.finished = False
+ self.cleanup_done = False
+ self.last_status = None
+ self.cleanup_on_reassignment = cleanup_on_reassignment
+ TriggerInstances().add_instances(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 should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ return not (
+ termination_reason == TriggerTerminationReason.REASSIGNED and not self.cleanup_on_reassignment
+ )
Review Comment:
```suggestion
return (
termination_reason != TriggerTerminationReason.REASSIGNED
or self.cleanup_on_reassignment
)
```
(Or is more difficult to understand?)
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +757,28 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
+ # find out reassigned triggers
+ reassigned_trigger_ids = set(Trigger.filter_out_reassigned_triggers(triggerer_id, cancel_trigger_ids))
+
+ other_reasons_trigger_ids = cancel_trigger_ids - reassigned_trigger_ids
+
+ trigger_id_reasons = add_reason(reassigned_trigger_ids, TriggerTerminationReason.REASSIGNED)
+ trigger_id_reasons.extend(add_reason(other_reasons_trigger_ids, TriggerTerminationReason.OTHER))
+
+ return trigger_id_reasons
Review Comment:
```suggestion
reassigned_pairs = (
(trigger_id, TriggerTerminationReason.REASSIGNED)
for trigger_id in reassigned_trigger_ids
)
other_reason_pairs = (
(trigger_id, TriggerTerminationReason.OTHER)
for trigger_id in other_reasons_trigger_ids
)
return [*reassigned_pairs, *other_reason_pairs]
```
Same as your code but easier to read IMO. The additional `add_reason` function doesn’t make things easier.
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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()
+
+ def add_instances(self, instance):
+ self.trigger_instances.append(instance)
+
+
+class RemoteJobTrigger(BaseTrigger):
+ def __init__(self, remote_job_id, cleanup_on_reassignment: bool):
+ super().__init__()
+ self.remote_job_id = remote_job_id
+ self.finished = False
+ self.cleanup_done = False
+ self.last_status = None
+ self.cleanup_on_reassignment = cleanup_on_reassignment
+ TriggerInstances().add_instances(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()}")
Review Comment:
These prints should be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716600418
##########
airflow/triggers/base.py:
##########
@@ -115,6 +116,19 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ def should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ """
+ Check the trigger should be cleaned up or not base on the context.
+
+ :param termination_reason: The reason for terminating the trigger.
+ Since the trigger could be terminated for various reasons, like triggerer restart or reassigned to
+ another triggerer, this method allows the trigger to decide whether it should be cleaned up under the
+ current circumstances.
+
+ By default, will always return True, override this method base custom requirements.
Review Comment:
Not for testing purpose, this docstring is a little confuse here. I’ve updated the description to indicate that, for backward compatibility, the method will always return True by default. Users can override this method based on their specific needs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717199660
##########
airflow/models/trigger.py:
##########
@@ -308,3 +308,21 @@ def get_sorted_triggers(cls, capacity: int, alive_triggerer_ids: list[int] | Sel
skip_locked=True,
)
return session.execute(query).all()
+
+ @classmethod
+ @internal_api_call
+ @provide_session
+ def filter_out_reassigned_triggers(
Review Comment:
function should be renamed. it's not filtering them out -- it's "getting them"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717193757
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +648,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details["termination_reason"]):
Review Comment:
Rather than adding a new method "should_cleanup" i think it would be better if we just add `reason` as an argument to `cleanup` signature. Then cleanup can choose to do nothing in that case. Simpler and more direct that way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716328637
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -538,10 +539,13 @@ async def cancel_triggers(self):
This allows the cleanup job to delete them.
"""
while self.to_cancel:
- trigger_id = self.to_cancel.popleft()
+ trigger_id, reason = self.to_cancel.popleft()
if trigger_id in self.triggers:
+ # Add more context here before cancelling
+ to_be_cancelled = self.triggers[trigger_id]
+ to_be_cancelled["termination_reason"] = reason
# We only delete if it did not exit already
- self.triggers[trigger_id]["task"].cancel()
+ to_be_cancelled["task"].cancel()
Review Comment:
Ah, it is the same object in `self.triggers` so the additional information is stored there. Nevermind.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-2288312445
Also, please consider to configure pre-commit locally to verify the code before you push. This will save a lot of time waiting for CI checking simple formatting errors.
You can find guides on this in `contributing-docs`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717836475
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +648,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details["termination_reason"]):
Review Comment:
Got it. I've already removed the should_cleanup and added the termination context to the cleanup method. The documentation has been updated as well. @dstandish
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-2292648881
Many thanks for the valuable suggestions from @uranusjr @dstandish ! Just a follow-up on the PR process. What are the next action items? Please let me know, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716607556
##########
airflow/triggers/base.py:
##########
@@ -115,6 +116,19 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ def should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ """
+ Check the trigger should be cleaned up or not base on the context.
+
+ :param termination_reason: The reason for terminating the trigger.
+ Since the trigger could be terminated for various reasons, like triggerer restart or reassigned to
+ another triggerer, this method allows the trigger to decide whether it should be cleaned up under the
+ current circumstances.
+
+ By default, will always return True, override this method base custom requirements.
Review Comment:
Some text should also be added to the documentation (probably in the *Deferrable Operators & Triggers* page?) to teach users when/how this can be used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-2268150420
CC @potiuk If there are any other suggested solutions for this, our use case is as follows:
1.Submit a job to Apache Kyuubi in the worker, which will return a remote job ID and the remote job executed in Kyuubi side.
2.Defer the task by creating a trigger and pass the remote job ID to the trigger.
3.The trigger will continuously check the status of the remote job in the run method and yield an event once the job succeeds or fails.
4.In the trigger's cleanup method, we terminate the remote job if the remote job ID exists.
The problem is that the cleanup method could called when a task is canceled or the trigger is reassigned to another triggerer. The issue arises when the cleanup method is called during reassignment, which terminates the remote job. As a result, once reassigned to another triggerer, the remote job fails.
To avoid making the trigger heavier, we chose not to move step 1 (submitting the job and receiving the remote job ID) into the trigger's run method. This decision is based on the need to handle many preparations, such as authentication setup and local file propagation, which could burden the trigger process. Instead, we perform these preparations in the worker and delegate only the status check to the trigger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717846006
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +756,31 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
Review Comment:
```suggestion
```
No longer used anywhere?
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +756,31 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
Review Comment:
It seems like this is only intended to be used as an internal method. Add a leading underscore to the name (`_add_trigger_cancel_reasons`) to indicate this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716592318
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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()
+
+ def add_instances(self, instance):
+ self.trigger_instances.append(instance)
+
+
+class RemoteJobTrigger(BaseTrigger):
+ def __init__(self, remote_job_id, cleanup_on_reassignment: bool):
+ super().__init__()
+ self.remote_job_id = remote_job_id
+ self.finished = False
+ self.cleanup_done = False
+ self.last_status = None
+ self.cleanup_on_reassignment = cleanup_on_reassignment
+ TriggerInstances().add_instances(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()}")
Review Comment:
Sure, removed the print now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717736519
##########
airflow/models/trigger.py:
##########
@@ -308,3 +308,21 @@ def get_sorted_triggers(cls, capacity: int, alive_triggerer_ids: list[int] | Sel
skip_locked=True,
)
return session.execute(query).all()
+
+ @classmethod
+ @internal_api_call
+ @provide_session
+ def filter_out_reassigned_triggers(
Review Comment:
Thanks, I already renamed to get_reassigned_triggers
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #41232: fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task
URL: https://github.com/apache/airflow/pull/41232
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716611828
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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:
Review Comment:
The purpose of this approach is that the trigger, once instantiated, isn't stored anywhere and is directly put into running state after initialization, making it impossible to access the instance directly. Therefore, I have made it register itself in a singleton during initialization, which allows me to access it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716588297
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -441,6 +441,7 @@ class TriggerDetails(TypedDict):
task: asyncio.Task
name: str
events: int
+ termination_reason: TriggerTerminationReason | None
Review Comment:
Thanks, add the initializing in create_triggers
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716621892
##########
airflow/triggers/base.py:
##########
@@ -115,6 +116,19 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ def should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ """
+ Check the trigger should be cleaned up or not base on the context.
+
+ :param termination_reason: The reason for terminating the trigger.
+ Since the trigger could be terminated for various reasons, like triggerer restart or reassigned to
+ another triggerer, this method allows the trigger to decide whether it should be cleaned up under the
+ current circumstances.
+
+ By default, will always return True, override this method base custom requirements.
Review Comment:
Yes, let me see how to add it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-3832249738
> This change has been in our internal Airflow build for two years and has proven effective at preventing unexpected termination of trigger jobs by capturing cleanup context in scenarios such as deadlocks, instance restarts (in Kubernetes environments), and trigger job reassignment. It seems the discussion is still ongoing, and I’d like to follow up. It would be great if this change could be accepted upstream so we don’t have to maintain it internally with each Airflow upgrade. cc @kaxil
Hi @TakawaAkirayo , could you create a new PR against main, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716331102
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -643,7 +647,9 @@ async def run_trigger(self, trigger_id, trigger):
# allow triggers a chance to cleanup, either in that case or if
# they exit cleanly. Exception from cleanup methods are ignored.
with suppress(Exception):
- await trigger.cleanup()
+ trigger_details = self.triggers[trigger_id]
+ if trigger.should_cleanup(trigger_details.get("termination_reason")):
Review Comment:
… and this should not use `get()` but just regular `[]`.
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -441,6 +441,7 @@ class TriggerDetails(TypedDict):
task: asyncio.Task
name: str
events: int
+ termination_reason: TriggerTerminationReason | None
Review Comment:
This key should be initialized in `create_triggers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716512066
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +757,28 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
+ # find out reassigned triggers
+ reassigned_trigger_ids = set(Trigger.filter_out_reassigned_triggers(triggerer_id, cancel_trigger_ids))
+
+ other_reasons_trigger_ids = cancel_trigger_ids - reassigned_trigger_ids
+
+ trigger_id_reasons = add_reason(reassigned_trigger_ids, TriggerTerminationReason.REASSIGNED)
+ trigger_id_reasons.extend(add_reason(other_reasons_trigger_ids, TriggerTerminationReason.OTHER))
+
+ return trigger_id_reasons
Review Comment:
I agree. I initially thought the reasons might be expanded here, but since there are currently only two types, let's adopt your suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716608438
##########
airflow/models/trigger.py:
##########
@@ -308,3 +308,21 @@ def get_sorted_triggers(cls, capacity: int, alive_triggerer_ids: list[int] | Sel
skip_locked=True,
)
return session.execute(query).all()
+
+ @classmethod
+ @internal_api_call
+ @provide_session
+ def filter_out_reassigned_triggers(
+ cls, triggerer_id: int, local_trigger_ids: set[int], session: Session = NEW_SESSION
+ ) -> set[int]:
+ """
+ Get the triggers that reassigned to other triggerers.
+
+ :param triggerer_id: Current triggerer's id.
+ :param local_trigger_ids: The ids of triggers that running in this triggerer.
+ :param session: The database session
+ """
+ reassigned_trigger_ids = session.scalars(
+ select(cls.id).where(cls.triggerer_id != triggerer_id, cls.id.in_(local_trigger_ids))
+ ).all()
Review Comment:
```suggestion
)
```
Don’t need this, `set()` below resolves the query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-3827580685
This change has been in our internal Airflow build for two years and has proven effective at preventing unexpected termination of trigger jobs by capturing cleanup context in scenarios such as deadlocks, instance restarts (in Kubernetes environments), and trigger job reassignment. It seems the discussion is still ongoing, and I’d like to follow up. It would be great if this change could be accepted upstream so we don’t have to maintain it internally with each Airflow upgrade. cc @kaxil
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-2266538936
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716531837
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -787,3 +801,323 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+
+
+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()
+
+ def add_instances(self, instance):
+ self.trigger_instances.append(instance)
+
+
+class RemoteJobTrigger(BaseTrigger):
+ def __init__(self, remote_job_id, cleanup_on_reassignment: bool):
+ super().__init__()
+ self.remote_job_id = remote_job_id
+ self.finished = False
+ self.cleanup_done = False
+ self.last_status = None
+ self.cleanup_on_reassignment = cleanup_on_reassignment
+ TriggerInstances().add_instances(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 should_cleanup(self, termination_reason: TriggerTerminationReason | None) -> bool:
+ return not (
+ termination_reason == TriggerTerminationReason.REASSIGNED and not self.cleanup_on_reassignment
+ )
Review Comment:
Logically, I wanted to emphasize that cleanup should not occur during reassignment, but this makes it appear cumbersome and unnecessary. So let's go with your suggestion—it makes the logic look clearer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1716590969
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -751,3 +757,28 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]:
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ @staticmethod
+ def add_trigger_cancel_reasons(
+ triggerer_id, cancel_trigger_ids: set[int]
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ """
+ Add trigger cancel reasons to give consumer more context.
+
+ Currently, we only distinguish between reassigned and other reasons.
+ """
+
+ def add_reason(
+ ids: set[int], reason: TriggerTerminationReason
+ ) -> list[tuple[int, TriggerTerminationReason]]:
+ return [(trigger_id, reason) for trigger_id in ids]
+
+ # find out reassigned triggers
+ reassigned_trigger_ids = set(Trigger.filter_out_reassigned_triggers(triggerer_id, cancel_trigger_ids))
Review Comment:
You are right, already changed the method to return set directly now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #41232:
URL: https://github.com/apache/airflow/pull/41232#discussion_r1717200270
##########
airflow/models/trigger.py:
##########
@@ -308,3 +308,21 @@ def get_sorted_triggers(cls, capacity: int, alive_triggerer_ids: list[int] | Sel
skip_locked=True,
)
return session.execute(query).all()
+
+ @classmethod
+ @internal_api_call
+ @provide_session
+ def filter_out_reassigned_triggers(
Review Comment:
i.e. it returns the list of reassigned triggers. something _else_ might filter out but this does not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: The remote task tracked by the trigger failed due to the reassignment of the trigger task [airflow]
Posted by "TakawaAkirayo (via GitHub)" <gi...@apache.org>.
TakawaAkirayo commented on PR #41232:
URL: https://github.com/apache/airflow/pull/41232#issuecomment-3833707046
> > This change has been in our internal Airflow build for two years and has proven effective at preventing unexpected termination of trigger jobs by capturing cleanup context in scenarios such as deadlocks, instance restarts (in Kubernetes environments), and trigger job reassignment. It seems the discussion is still ongoing, and I’d like to follow up. It would be great if this change could be accepted upstream so we don’t have to maintain it internally with each Airflow upgrade. cc @kaxil
>
> Hi @TakawaAkirayo , could you create a new PR against main, please?
Sure, the main now is Airflow 3, let me take some times to apply to it and verify
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org