Add deferrable mode to ExternalTaskSensor #29260

bkossakowska · 2023-01-31T07:54:04Z

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
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.

airflow/sensors/external_task.py

kaxil · 2023-02-02T13:40:05Z

airflow/triggers/external_task.py

+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations


I have added Triggers for external_task, can you please rebase on Main.

For sensors, I would be grateful if you can help the community by maintaining compatibility with astronomer-providers -- There are community members as well as our users using it https://pypistats.org/packages/astronomer-providers

Thank you for great implementation os the trigger, i have updated the sensor to use it.
However, after some testing i have found some moments that could be improved in the trigger:

There is no possibility to process several external tasks at once, using external_task_ids parameter. For that reason ExternalTaskSensor now supports only processing one task per run. Can you please take a look on it and add the possibility?;

If you pass name of the Dag that does not exist at all, the trigger will run infinitely trying to query task from that Dag. I have added one additional parameter to set a deadline for this search to prevent infinitive loop. If it will not find specific Dag for a minute, it will terminate.

airflow/sensors/external_task.py

VladaZakharova · 2023-02-07T15:25:17Z

Hi @kaxil !
Please, check the changes and let me know if there is something that i can improve. Thanks!

kaxil

This lgtm, some minor nits. Could you add unit tests to the Async Sensor please similar to the ones in https://github.com/astronomer/astronomer-providers/blob/main/tests/core/sensors/test_external_task.py

airflow/example_dags/example_external_task_marker_dag.py

kaxil · 2023-02-08T11:01:13Z

airflow/triggers/external_task.py

+                        await asyncio.sleep(self.poll_interval)
+                else:
+                    yield TriggerEvent({"status": "timeout"})
+                    return


Don't think we need return here

If there is no return statement then trigger logic won't work in python 3.7. Trigger will do it's logic of checking task status infinitely and won't stop even after yielding the event. Also source code of the base class says that it is required to return None after yield statement.
I couldn't find this in the documentation, but I found it by myself while was implementing trigger for python3.7. I think that return statement should be there so the interpreter would exit from the generator loop when specified event appears else we get an endless loop.

kaxil · 2023-02-08T11:01:20Z

airflow/triggers/external_task.py

+                    return
+                if await self.count_tasks() == len(self.execution_dates):
+                    yield TriggerEvent({"status": "success"})
+                    return


same as above

kaxil · 2023-02-08T11:01:25Z

airflow/triggers/external_task.py

+                await asyncio.sleep(self.poll_interval)
+            except Exception:
+                yield TriggerEvent({"status": "failed"})
+                return


same as above

VladaZakharova · 2023-02-17T14:59:27Z

Hi @kaxil !
Is there something else i can improve in the code?

potiuk · 2023-02-24T09:57:38Z

LGTM @kaxil ?

github-actions · 2023-04-11T00:11:38Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

tanvn · 2023-04-11T01:05:45Z

Just a gentle reminder to @kaxil
As a heavy user of Airflow, I would love to try this feature soon.

codecae · 2023-04-12T16:50:52Z

Just a gentle reminder to @kaxil As a heavy user of Airflow, I would love to try this feature soon.

Same! :)

airflow/sensors/external_task.py

kaxil · 2023-05-11T00:57:40Z

lgtm. Any final thoughts @phanikumv @pankajastro ?

phanikumv

Please add documentation about the new functionality here in /docs/apache-airflow/howto/operator/external_task_sensor.rst

airflow/sensors/external_task.py

tests/system/providers/core/example_external_task_parent_deferrable.py

pankajastro

Hi @bkossakowska, would it make sense to add deferrable in ExternalTaskSensor and not create a new sensor? I know at some places we have two sensors/operators one for sync and the other for async but at most of the places we have a deferrable param which controls this behaviour.

bkossakowska · 2023-05-18T10:31:39Z

Hi @pankajastro, @phanikumv,
thanks for your comments, changes have been added. Is there anything else I can improve?

OfSixes · 2023-07-06T12:05:17Z

airflow/triggers/external_task.py

@@ -38,25 +39,33 @@ class TaskStateTrigger(BaseTrigger):
    :param task_id: The task_id that contains the task you want to
        wait for. If ``None`` (default value) the sensor waits for the DAG


The statement "If None the sensor waits for the DAG" is untrue, no such behavior is currently implemented in this Trigger

OfSixes · 2023-07-06T12:06:03Z

airflow/triggers/external_task.py

@@ -38,25 +39,33 @@ class TaskStateTrigger(BaseTrigger):
    :param task_id: The task_id that contains the task you want to
        wait for. If ``None`` (default value) the sensor waits for the DAG
    :param states: allowed states, default is ``['success']``


While having the default value be ["success"] would make a lot of sense, no default value is currently set for the states parameter.

Currently, for the states parameter, as it is in the description, we have set it to ["success"] by default.
list(allowed_states) if allowed_states else [TaskInstanceState.SUCCESS.value].

This default is set in the sensor though, in the trigger itself this behavior is not implemented. If someone now wanted to use the trigger seperately, the documentation for it is incorrect.

VladaZakharova · 2023-07-13T07:47:45Z

Hi @potiuk !
Could you please check the changes? thanks :)

boring-cyborg bot added area:core-operators Operators, Sensors and hooks within Core Airflow area:system-tests labels Jan 31, 2023

kaxil reviewed Jan 31, 2023

View reviewed changes

airflow/sensors/external_task.py Outdated Show resolved Hide resolved

kaxil reviewed Feb 2, 2023

View reviewed changes

airflow/sensors/external_task.py Outdated Show resolved Hide resolved

VladaZakharova force-pushed the sensor_def_mode branch from 363bd06 to 27d5a7e Compare February 3, 2023 14:47

eladkal modified the milestones: Airflow 2.5.2, Airflow 2.6.0 Feb 4, 2023

kaxil reviewed Feb 8, 2023

View reviewed changes

potiuk approved these changes Feb 24, 2023

View reviewed changes

Taragolis mentioned this pull request Mar 14, 2023

Add support for deferrable operators in AMPP #30032

Merged

github-actions bot added the stale Stale PRs per the .github/workflows/stale.yml policy file label Apr 11, 2023

github-actions bot removed the stale Stale PRs per the .github/workflows/stale.yml policy file label Apr 12, 2023

ephraimbuddy modified the milestones: Airflow 2.6.0, Airflow 2.6.1 Apr 13, 2023

ashb modified the milestones: Airflow 2.6.1, Airflow 2.7.0 May 6, 2023

bkossakowska closed this May 10, 2023

bkossakowska reopened this May 10, 2023

kaxil approved these changes May 11, 2023

View reviewed changes

kaxil reviewed May 11, 2023

View reviewed changes

airflow/sensors/external_task.py Outdated Show resolved Hide resolved

phanikumv requested changes May 11, 2023

View reviewed changes

airflow/sensors/external_task.py Outdated Show resolved Hide resolved

phanikumv reviewed May 11, 2023

View reviewed changes

tests/system/providers/core/example_external_task_parent_deferrable.py Show resolved Hide resolved

pankajastro reviewed May 11, 2023

View reviewed changes

bkossakowska force-pushed the sensor_def_mode branch 3 times, most recently from 0af46d1 to e88ff7e Compare May 18, 2023 10:18

phanikumv approved these changes May 18, 2023

View reviewed changes

bkossakowska force-pushed the sensor_def_mode branch from e88ff7e to 54d6c58 Compare June 1, 2023 08:55

OfSixes reviewed Jul 6, 2023

View reviewed changes

Beata Kossakowska and others added 11 commits July 12, 2023 08:20

Add deferrable mode to ExternalTaskSensor

cb5a7b7

Rebase main and change ExternalTaskAsyncSensor implementation

f88e363

Update __init__.py

4f6040f

Add unit tests for the AsyncSensor

bee2d6c

Update message

2d32bc6

Update message

99bf460

Update message

9bc8d5c

Update test message

226ff69

update test

d2987bd

Merge ExternalTaskAsyncSensor into ExternalTaskSensor

994aa43

Fix static checks

e801b7d

bkossakowska force-pushed the sensor_def_mode branch 2 times, most recently from 38a4158 to c6537a5 Compare July 12, 2023 09:39

Rebase main and fix.

9c314d1

bkossakowska force-pushed the sensor_def_mode branch from c6537a5 to 9c314d1 Compare July 12, 2023 10:58

potiuk approved these changes Jul 20, 2023

View reviewed changes

potiuk merged commit dda3dcd into apache:main Jul 20, 2023

ephraimbuddy added the changelog:skip Changes that should be skipped from the changelog (CI, tests, etc..) label Aug 2, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add deferrable mode to ExternalTaskSensor #29260

Add deferrable mode to ExternalTaskSensor #29260

bkossakowska commented Jan 31, 2023

kaxil Feb 2, 2023

VladaZakharova Feb 3, 2023

VladaZakharova commented Feb 7, 2023

kaxil left a comment

kaxil Feb 8, 2023

VladaZakharova Feb 8, 2023

kaxil Feb 8, 2023

kaxil Feb 8, 2023

VladaZakharova commented Feb 17, 2023

potiuk commented Feb 24, 2023

github-actions bot commented Apr 11, 2023

tanvn commented Apr 11, 2023

codecae commented Apr 12, 2023

kaxil commented May 11, 2023 •

edited

Loading

phanikumv left a comment •

edited

Loading

pankajastro left a comment •

edited

Loading

bkossakowska commented May 18, 2023

OfSixes Jul 6, 2023

OfSixes Jul 6, 2023 •

edited

Loading

bkossakowska Jul 12, 2023

OfSixes Jul 12, 2023

VladaZakharova commented Jul 13, 2023

		@@ -38,25 +39,33 @@ class TaskStateTrigger(BaseTrigger):
		:param task_id: The task_id that contains the task you want to
		wait for. If ``None`` (default value) the sensor waits for the DAG

Add deferrable mode to ExternalTaskSensor #29260

Add deferrable mode to ExternalTaskSensor #29260

Conversation

bkossakowska commented Jan 31, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VladaZakharova commented Feb 7, 2023

kaxil left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VladaZakharova commented Feb 17, 2023

potiuk commented Feb 24, 2023

github-actions bot commented Apr 11, 2023

tanvn commented Apr 11, 2023

codecae commented Apr 12, 2023

kaxil commented May 11, 2023 • edited Loading

phanikumv left a comment • edited Loading

Choose a reason for hiding this comment

pankajastro left a comment • edited Loading

Choose a reason for hiding this comment

bkossakowska commented May 18, 2023

Choose a reason for hiding this comment

OfSixes Jul 6, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VladaZakharova commented Jul 13, 2023

kaxil commented May 11, 2023 •

edited

Loading

phanikumv left a comment •

edited

Loading

pankajastro left a comment •

edited

Loading

OfSixes Jul 6, 2023 •

edited

Loading