Fix DateTimeSensorAsync crash on templated target_time with start_from_trigger (real fix, closes #70284) - #70576
Fix DateTimeSensorAsync crash on templated target_time with start_from_trigger (real fix, closes #70284)#70576nanjeshramesh wants to merge 10 commits into
Conversation
…m_trigger Real fix for apache#70284 (not just a nicer error message like apache#70310/apache#70469 -- see PR description). Defers target_time resolution to the triggerer via template-field rendering, same mechanism FileSensor uses for filepath (apache#55068).
|
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
|
SameerMesiah97
left a comment
SameerMesiah97
left a comment
There was a problem hiding this comment.
Looks good to me overall. But I would go through the diff again and see if any docstrings/comments could be shortened/removed. I have pointed a few of them that I felt were a bit too verbose but there are certainly others that can be more concise.
CI needs to be triggered
| # target_time couldn't be parsed as a static datetime at Dag-parse time. This is | ||
| # the normal, documented case of target_time being a Jinja template, e.g. | ||
| # "{{ data_interval_end.tomorrow().replace(hour=1) }}" -- not necessarily bad input. | ||
| moment = None |
There was a problem hiding this comment.
This catches all ValueErrors and treats them as if the value were simply "not yet renderable". Is that intentional? For example, an invalid static value like "not-a-date" would now follow the same path as an unresolved Jinja template. Should those cases be distinguished so genuinely invalid input still fails eagerly?
| This requires either a static ``target_time`` (a datetime or ISO-8601 string) or, on | ||
| Airflow >= 3.3, a templated ``target_time`` that the triggerer can render before the trigger | ||
| runs. On earlier Airflow versions a templated ``target_time`` cannot be resolved this way, so | ||
| ``start_from_trigger`` is disabled with a warning and the task defers from the worker instead. |
There was a problem hiding this comment.
This can be more concise. Please see the below:
:param start_from_trigger: Start the task directly from the triggerer instead of a worker.
Supports static ``target_time`` values and, on Airflow >= 3.3, templated
``target_time`` values.
| # operator's template field ("target_time"). The triggerer renders any | ||
| # start_trigger_args kwarg whose name matches an operator template field before | ||
| # running the trigger (see BaseTrigger.task_instance / render_template_fields), | ||
| # the same mechanism FileSensor relies on for a templated `filepath`. |
There was a problem hiding this comment.
The comment is too long. Please see the below;
# Pass the unresolved template to the trigger. On Airflow >= 3.3 the
# triggerer renders trigger kwargs that correspond to template fields before
# starting the trigger.
| # runs, so a templated target_time can never be resolved via start_from_trigger. | ||
| # Falling back to the normal worker-deferred path (execute()) avoids crashing Dag | ||
| # parsing on every parse cycle; execute() runs after the scheduler/worker has | ||
| # already rendered target_time normally. |
There was a problem hiding this comment.
Is this comment needed? Looks like the warning message communicates the necessary information. I think this comment can be removed or trimmed.
| ``template_fields``, so the triggerer renders it in place (see | ||
| ``BaseTrigger.render_template_fields``) before ``run()`` is invoked; it is then parsed into | ||
| ``moment`` on first use. This mirrors how ``FileSensor`` defers rendering of a templated | ||
| ``filepath`` to the triggerer. |
There was a problem hiding this comment.
This is too long as well. Please see the below:
:param target_time: Templated ``target_time`` value to resolve when the trigger
runs. Used instead of ``moment`` when the target time cannot be determined
during operator initialization. Mutually exclusive with ``moment``.
| if self.moment is not None: | ||
| return self.moment | ||
| if not self.target_time: | ||
| raise TypeError("DateTimeTrigger requires either 'moment' or 'target_time' to be set") |
There was a problem hiding this comment.
Why duplicate the validation logic in the constructor here? Will it not be guaranteed that moment or target_time is set by the time _resolve_moment is called?
| # scripts/ci/prek/check_trigger_serialize_init.py). By the time serialize() is called, | ||
| # target_time has either already been rendered by the triggerer (see | ||
| # BaseTrigger.render_template_fields) or resolution raises a clear error -- there is no | ||
| # unrendered template left to preserve. |
There was a problem hiding this comment.
Not sure if this comment is needed. The code is self-explanatory.
| dag=self.dag, | ||
| ) | ||
| assert op.start_from_trigger is True | ||
| assert op.start_trigger_args.trigger_kwargs["moment"] == pendulum.parse("2020-01-01T00:00:00+00:00") |
There was a problem hiding this comment.
Would it be worth adding a test for a non-templated bt invalid target_time (e.g. "not-a-date") so the expected behaviour for genuinely invalid input is explicit?
|
Thanks for the thorough review! On the On Docstrings/comments — trimmed all of them to roughly your suggested wording (the Pushed as three commits: fail-fast fix + trimmed docs, further comment trims, and the new test. |
2 participants
DateTimeSensorAsync.__init__callstimezone.parse(self.target_time)at Dag-parse timewhen
start_from_trigger=True, before Jinja rendering has happened.target_timeis adocumented template field (e.g.
"{{ data_interval_end.tomorrow().replace(hour=1) }}"), sopassing a template — the field's normal, documented use case — raises a raw
pendulum.parsing.exceptions.ParserErrorright there, crashing parsing of the entire Dagfile, not just the one task.
Relation to #70310 and #70469
Both prior attempts at this issue (#70310 by @haseebmalik18, #70469 by @burakcoleman) catch
the parse failure in
__init__and raise a clearerValueErrorinstead of the rawParserError. That's a real improvement to the error message, but as reviewers on both PRspointed out, it doesn't fix the crash — the Dag still fails to parse whenever
start_from_trigger=Trueandtarget_timeis a template.This PR takes the approach @haseebmalik18 outlined in the #70310 review discussion: defer
resolving
target_timeuntil the triggerer can render it, the same wayFileSensoralreadyhandles a templated
filepath.What this PR does
On Airflow >= 3.3, instead of parsing
target_timeat Dag-parse time, the raw (unrendered)template string is passed straight through to
DateTimeTriggerviastart_trigger_args.trigger_kwargsunder the key"target_time"— which matchesDateTimeSensor.template_fields.BaseTrigger'stask_instancesetter already picks up anytrigger_kwargskey that matches an operator template field (seeairflow.triggers.base.BaseTrigger), andTriggererJobRunner.run_triggercallsrender_template_fields()on the trigger before invokingrun()— this is exactly themechanism introduced in #55068 and that
FileSensoralready relies on forfilepath.DateTimeTriggernow accepts either a resolvedmoment(unchanged, existing behavior) or araw
target_timestring. In the latter case,momentstarts asNoneand is lazily parsedfrom
target_timethe first time it's needed (run()orserialize()), by which point thetriggerer has already rendered the template in place.
On Airflow < 3.3, that triggerer-side rendering mechanism doesn't exist yet, so a templated
target_timecan never be resolved viastart_from_trigger. Rather than crash Dag parsing onevery parse cycle,
start_from_triggeris disabled with alog.warning, and the task fallsback to deferring from the worker via
execute()instead — at that pointtarget_timehasalready been rendered normally by the scheduler/worker, so it works correctly, just without
the trigger-only optimization.
Static/ISO
target_timevalues are completely unaffected on any version.Testing
test_temporal.pycovering:DateTimeTriggeraccepting an unrenderedtarget_timeat construction without raising, resolving it once rendered, raising a clearerror if it's still unrendered when needed, rejecting both/neither of
moment/target_time,and a full simulated triggerer flow (construct with raw template →
render_template_fields→
run()) that fires at the correct time.test_date_time.pycoveringDateTimeSensorAsync.__init__branching onAIRFLOW_V_3_3_PLUSfor a templatedtarget_time, and confirming a statictarget_timebehaves identically regardless of that flag.
pendulum.parsing.exceptions.ParserErroronmain, and parses cleanly with this fix.ruff check/ruff format --diffclean on all changed files.closes: #70284
Related: #70310, #70469, #55068, #69610
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Sonnet 5 following the guidelines