Under which category would you file this issue?
Airflow Core
Apache Airflow version
3.3.0+
What happened and how to reproduce it?
The evaluation-timing category passed to @deadline_reference is never persisted, and the one place that needs it doesn't actually use it, relying on identifying queued-anchored deadlines by hardcoded class name instead. Custom references are therefore silently excluded.
Clearing a DagRun re-queues it and refreshes queued_at (DagRun.set_state), and _recalculate_dagrun_queued_at_deadlines in models/taskinstance.py re-anchors the affected deadlines. Its selection criterion is a string comparison (line 244):
DeadlineAlertModel.reference[ReferenceModels.REFERENCE_TYPE_FIELD].as_string()
== ReferenceModels.DagRunQueuedAtDeadline.__name__,
But reference_type holds the reference's own bare __name__, because
BaseDeadlineReference.reference_name returns self.__class__.__name__ and the base serialize_reference emits {REFERENCE_TYPE_FIELD: self.reference_name}. Confirmed against stored rows:
reference_type | class_path
--------------------------+---------------------------------------
DagRunQueuedAtDeadline | <- built-in, matches
DagRunLogicalDateDeadline |
CloseOfBusinessDeadline | cob_reference.CloseOfBusinessDeadline <- custom, never matches
So a custom reference decorated @deadline_reference(DeadlineReference.TYPES.DAGRUN_QUEUED) can never satisfy that filter, no matter what it was registered as. register_custom_reference appends the class to the in-memory TYPES.DAGRUN_QUEUED tuple in whichever process ran the Dag file, and nothing writes that category to DeadlineAlertModel, so the row the query reads has no timing field to filter on even in principle.
To reproduce:
- Register a custom reference with
@deadline_reference(DeadlineReference.TYPES.DAGRUN_QUEUED) whose _evaluate_with returns queued_at plus something.
- Put it on one Dag and
DeadlineReference.DAGRUN_QUEUED_AT on another.
- Trigger both, record
deadline.deadline_time.
- Clear both runs, and compare. The built-in is re-anchored to the new
queued_at; the custom one is not.
What you think should happen instead?
The registered timing category should determine which deadlines are re-anchored, so that a custom queued-anchored reference behaves like the built-in one.
- The
DeadlineReference.TYPES value needs to be added to the serialized reference, and
- the check in
_recalculate_dagrun_queued_at_deadlines which currently says
DeadlineAlertModel.reference[ReferenceModels.REFERENCE_TYPE_FIELD].as_string()
== ReferenceModels.DagRunQueuedAtDeadline.__name__,
needs to check that value instead of the __name__.
Operating System
No response
Deployment
None
Apache Airflow Provider(s)
No response
Versions of Apache Airflow Providers
No response
Official Helm Chart version
Not Applicable
Kubernetes Version
No response
Helm Chart configuration
No response
Docker Image customizations
No response
Anything else?
No response
Are you willing to submit PR?
Code of Conduct
Under which category would you file this issue?
Airflow Core
Apache Airflow version
3.3.0+
What happened and how to reproduce it?
The evaluation-timing category passed to
@deadline_referenceis never persisted, and the one place that needs it doesn't actually use it, relying on identifying queued-anchored deadlines by hardcoded class name instead. Custom references are therefore silently excluded.Clearing a DagRun re-queues it and refreshes
queued_at(DagRun.set_state), and_recalculate_dagrun_queued_at_deadlinesinmodels/taskinstance.pyre-anchors the affected deadlines. Its selection criterion is a string comparison (line 244):But
reference_typeholds the reference's own bare__name__, becauseBaseDeadlineReference.reference_namereturnsself.__class__.__name__and the baseserialize_referenceemits{REFERENCE_TYPE_FIELD: self.reference_name}. Confirmed against stored rows:So a custom reference decorated
@deadline_reference(DeadlineReference.TYPES.DAGRUN_QUEUED)can never satisfy that filter, no matter what it was registered as.register_custom_referenceappends the class to the in-memoryTYPES.DAGRUN_QUEUEDtuple in whichever process ran the Dag file, and nothing writes that category toDeadlineAlertModel, so the row the query reads has no timing field to filter on even in principle.To reproduce:
@deadline_reference(DeadlineReference.TYPES.DAGRUN_QUEUED)whose_evaluate_withreturnsqueued_atplus something.DeadlineReference.DAGRUN_QUEUED_ATon another.deadline.deadline_time.queued_at; the custom one is not.What you think should happen instead?
The registered timing category should determine which deadlines are re-anchored, so that a custom queued-anchored reference behaves like the built-in one.
DeadlineReference.TYPESvalue needs to be added to the serialized reference, and_recalculate_dagrun_queued_at_deadlineswhich currently saysneeds to check that value instead of the
__name__.Operating System
No response
Deployment
None
Apache Airflow Provider(s)
No response
Versions of Apache Airflow Providers
No response
Official Helm Chart version
Not Applicable
Kubernetes Version
No response
Helm Chart configuration
No response
Docker Image customizations
No response
Anything else?
No response
Are you willing to submit PR?
Code of Conduct