potiuk commented on code in PR #72475:
URL: https://github.com/apache/airflow/pull/72475#discussion_r4072545001
##########
airflow-core/src/airflow/timetables/_cron.py:
##########
@@ -61,15 +62,54 @@ def _covers_every_hour(cron: croniter) -> bool:
class CronMixin:
- """Mixin to provide interface to work with croniter."""
+ """
+ Mixin to provide interface to work with croniter.
+
+ Optionally applies a deterministic, per-DAG jitter to every scheduled time.
+ When ``max_jitter`` is set, each cron boundary is shifted by a fixed offset
+ derived from ``seed`` and spread across ``[0, max_jitter)``. This spreads
out
+ DAGs that share a cron expression (e.g. every ``@daily`` DAG firing at
+ midnight) so they no longer all fire at the same instant. The offset is
stable
+ for a given seed, so runs stay predictable across scheduler restarts and
+ serialization.
+
+ The offset shifts every cron-derived time uniformly. For data-interval
+ timetables this means the whole interval moves by the offset (the window
keeps
+ its length and consecutive intervals stay contiguous), not just the fire
time.
+
+ :param cron: cron expression (or a preset such as ``@daily``) defining the
schedule.
+ :param timezone: timezone used to interpret the cron expression.
+ :param seed: stable, unique-per-DAG string the offset is derived from; the
DAG id
+ is a natural choice. Must be non-empty whenever ``max_jitter`` is set.
+ :param max_jitter: upper bound of the jitter window; the offset falls in
+ ``[0, max_jitter)``. Defaults to zero, i.e. no jitter. Keep it small
relative
+ to the gap between cron boundaries.
+ """
- def __init__(self, cron: str, timezone: str | Timezone | FixedTimezone) ->
None:
+ def __init__(
+ self,
+ cron: str,
+ timezone: str | Timezone | FixedTimezone,
+ *,
+ seed: str = "",
+ max_jitter: datetime.timedelta = datetime.timedelta(),
+ ) -> None:
self._expression = cron_presets.get(cron, cron)
if isinstance(timezone, str):
timezone = parse_timezone(timezone)
self._timezone = timezone
+ if max_jitter > datetime.timedelta(0) and not seed:
+ raise ValueError("seed must be a non-empty, unique-per-DAG string
when max_jitter > 0")
Review Comment:
`minor` — a negative `max_jitter` slips through both guards: this check is
`max_jitter > timedelta(0)`, and the offset branch below is `if max_jitter_us >
0`, so the offset ends up zero. But `if self._max_jitter:` in `serialize()` is
truthy for a negative timedelta, so the payload still gets `"max_jitter":
-3600.0` and `"seed": ""`. Silently doing nothing while persisting a
nonsensical value — worth rejecting `max_jitter < 0` outright here.
##########
airflow-core/src/airflow/timetables/_cron.py:
##########
@@ -61,15 +62,54 @@ def _covers_every_hour(cron: croniter) -> bool:
class CronMixin:
- """Mixin to provide interface to work with croniter."""
+ """
+ Mixin to provide interface to work with croniter.
+
+ Optionally applies a deterministic, per-DAG jitter to every scheduled time.
+ When ``max_jitter`` is set, each cron boundary is shifted by a fixed offset
+ derived from ``seed`` and spread across ``[0, max_jitter)``. This spreads
out
+ DAGs that share a cron expression (e.g. every ``@daily`` DAG firing at
+ midnight) so they no longer all fire at the same instant. The offset is
stable
+ for a given seed, so runs stay predictable across scheduler restarts and
+ serialization.
+
+ The offset shifts every cron-derived time uniformly. For data-interval
+ timetables this means the whole interval moves by the offset (the window
keeps
+ its length and consecutive intervals stay contiguous), not just the fire
time.
+
+ :param cron: cron expression (or a preset such as ``@daily``) defining the
schedule.
+ :param timezone: timezone used to interpret the cron expression.
+ :param seed: stable, unique-per-DAG string the offset is derived from; the
DAG id
+ is a natural choice. Must be non-empty whenever ``max_jitter`` is set.
+ :param max_jitter: upper bound of the jitter window; the offset falls in
+ ``[0, max_jitter)``. Defaults to zero, i.e. no jitter. Keep it small
relative
+ to the gap between cron boundaries.
+ """
- def __init__(self, cron: str, timezone: str | Timezone | FixedTimezone) ->
None:
+ def __init__(
+ self,
+ cron: str,
+ timezone: str | Timezone | FixedTimezone,
+ *,
+ seed: str = "",
+ max_jitter: datetime.timedelta = datetime.timedelta(),
+ ) -> None:
self._expression = cron_presets.get(cron, cron)
if isinstance(timezone, str):
timezone = parse_timezone(timezone)
self._timezone = timezone
+ if max_jitter > datetime.timedelta(0) and not seed:
+ raise ValueError("seed must be a non-empty, unique-per-DAG string
when max_jitter > 0")
+ h = int(md5(seed.encode()).hexdigest(), 16)
Review Comment:
`nit` — this runs on every `CronMixin` construction, including the default
no-jitter path, i.e. for every cron Dag on every deserialization in the
scheduler. It is cheap, but it only matters when `max_jitter_us > 0`, so it
could move inside the branch below.
##########
airflow-core/src/airflow/timetables/trigger.py:
##########
@@ -406,8 +441,12 @@ def __init__(
run_offset: int | datetime.timedelta | relativedelta | None = None,
run_immediately: bool | datetime.timedelta = False,
key_format: str = r"%Y-%m-%dT%H:%M:%S",
+ seed: str = "",
+ max_jitter: datetime.timedelta = datetime.timedelta(),
Review Comment:
**Partition keys absorb the jitter here.**
`CronPartitionTimetable` inherits the shifted `_get_next`/`_align_to_next`,
and `_get_partition_date()` returns `run_date` unchanged for `run_offset == 0`,
so `_format_key()` formats the *jittered* instant. Using this PR's own offset
function:
```text
seed="my_dag", max_jitter=1h -> offset 0:58:51.663322
cron "0 0 * * *" -> key 2026-03-06T00:58:51 (was
2026-03-06T00:00:00)
seed="dag_4", max_jitter=2h -> offset 1:03:38.886020
cron "0 23 * * *", key_format="%Y-%m-%d"
-> key 2026-03-07 (was 2026-03-06)
```
Jitter should move *when* a run executes, not *which period it is for*. The
partition key is the run's identity — `iter_partition_dagrun_infos` dedups
backfills on it and it usually maps to a storage path — so switching jitter on
for an existing Dag renumbers every partition and stops it deduping against the
old ones. The second case shows the calendar date itself flipping once
`key_format` is coarser than the offset.
Suggest stripping the offset before the partition date is derived, so the
key stays on the boundary while the run still fires late. If the shifted key is
intended, it needs a line in the new docs section and a test — the
`CronPartitionTimetable` tests currently only assert the serialize round-trip.
##########
airflow-core/src/airflow/timetables/_cron.py:
##########
@@ -120,18 +168,26 @@ def _describe_with_dom_dow_fix(self, expression: str) ->
str:
def __eq__(self, other: object) -> bool:
"""
- Both expression and timezone should match.
+ Expression, timezone and jitter settings (``seed`` and ``max_jitter``)
should all match.
+
+ Two timetables that share a cron expression and timezone but differ in
jitter
+ produce different schedules, so they are not considered equal.
This is only for testing purposes and should not be relied on
otherwise.
"""
from airflow.serialization.encoders import coerce_to_core_timetable
if not isinstance(other := coerce_to_core_timetable(other),
type(self)):
return NotImplemented
- return self._expression == other._expression and self._timezone ==
other._timezone
+ return (
+ self._expression == other._expression
+ and self._timezone == other._timezone
+ and self._seed == other._seed
+ and self._max_jitter == other._max_jitter
+ )
def __hash__(self):
- return hash((self._expression, self._timezone))
+ return hash((self._expression, str(self._timezone), self._seed,
self._max_jitter))
Review Comment:
Confirmed this is a real latent bug, not a theoretical one — on current
`main`, `hash(CronTriggerTimetable("0 0 * * *", timezone="UTC"))` raises
`TypeError: unhashable type: 'Timezone'`. Good catch.
Since it is independent of the jitter feature and backportable on its own,
consider pulling it into a separate PR so it can land without waiting on this
one.
##########
task-sdk/src/airflow/sdk/definitions/timetables/_cron.py:
##########
@@ -41,16 +42,33 @@
@attrs.define
class CronMixin:
- """Mixin to provide interface to work with croniter."""
+ """
+ Mixin to provide interface to work with croniter.
+
+ Optionally applies a deterministic, per-DAG jitter: when ``max_jitter`` is
set, every
+ cron boundary is shifted by a fixed offset derived from ``seed`` and
spread across
+ ``[0, max_jitter)``, so DAGs sharing a cron expression no longer all fire
at the same
+ instant. The offset is computed scheduler-side; this class only carries
and validates
+ the settings.
+
+ :param seed: stable, unique-per-DAG string the offset is derived from (the
DAG id is a
+ natural choice). Must be non-empty whenever ``max_jitter`` is set.
+ :param max_jitter: upper bound of the jitter window; the offset falls in
+ ``[0, max_jitter)``. Defaults to zero, i.e. no jitter.
+ """
expression: str
timezone: str | Timezone | FixedTimezone
+ seed: str = attrs.field(kw_only=True, default="")
+ max_jitter: datetime.timedelta = attrs.field(kw_only=True,
default=datetime.timedelta())
def __attrs_post_init__(self) -> None:
# Resolve preset aliases (e.g. "@quarterly") to their cron expressions
# in-place. After this point the original preset string is lost;
# attrs.evolve, equality, and serialisation all see the resolved form.
self.expression = CRON_PRESETS.get(self.expression, self.expression)
+ if self.max_jitter > datetime.timedelta(0) and not self.seed:
+ raise ValueError("seed must be a non-empty, unique-per-DAG string
when max_jitter > 0")
Review Comment:
`minor` — this guard is only exercised by
`test_empty_seed_requires_zero_jitter` in
`airflow-core/tests/unit/timetables/`, so `breeze testing task-sdk-tests` —
which runs the Task SDK distribution standalone — does not cover it.
`task-sdk/tests/task_sdk/definitions/timetables/test__cron.py` already exists
and is the natural home for it.
##########
airflow-core/src/airflow/timetables/_cron.py:
##########
@@ -61,15 +62,54 @@ def _covers_every_hour(cron: croniter) -> bool:
class CronMixin:
- """Mixin to provide interface to work with croniter."""
+ """
+ Mixin to provide interface to work with croniter.
+
+ Optionally applies a deterministic, per-DAG jitter to every scheduled time.
Review Comment:
`minor` — `AGENTS.md` asks for `Dag` in prose:
> Write **Dag** (title case) in all prose. Keep the all-caps or lowercase
spelling only when reproducing a literal code token
This docstring has "per-DAG jitter", "every `@daily` DAG", "the DAG id is a
natural choice"; the same applies to the `:param seed:` docs in `trigger.py`,
both task-sdk files, the test docstrings, and the `ValueError` message
("unique-per-DAG string"). The `.rst` section added in this PR already uses
`Dag`, so it is just inconsistent within the PR.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
