You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Pebble32 (via GitHub)" <gi...@apache.org> on 2026/09/03 12:30:51 UTC
[PR] Add opt-in deterministic jitter to cron-based timetables [airflow]
Pebble32 opened a new pull request, #72475:
URL: https://github.com/apache/airflow/pull/72475
Add opt-in, deterministic jitter to every cron-based timetable via `CronMixin`: two kw-only params, `seed` and `max_jitter`, on `CronTriggerTimetable`, `CronDataIntervalTimetable`, `MultipleCronTriggerTimetable` and `CronPartitionTimetable`, in both the Task SDK and airflow-core, with serialization wired through. Each DAG's runs are shifted by a fixed offset drawn from `[0, max_jitter)`, so DAGs that share a cron expression no longer all fire at the same instant.
This is the follow-up @uranusjr suggested on #69705 (https://github.com/apache/airflow/pull/69705#issuecomment-5232872059): rather than a standalone timetable, the jitter lives in `CronMixin` so all cron scheduling inherits it. It supersedes #69705, which I will close once this lands. cc @kaxil, who reviewed the original.
### Why
`@daily` expands to `0 0 * * *`, so **every** daily DAG in a deployment is scheduled at exactly midnight, and the same holds for any shared cron expression. The existing ways to deal with this don't actually spread the schedule:
| Existing option | Why it doesn't spread the schedule |
| --- | --- |
| Hand-pick a unique minute per DAG | Manual, drifts and re-collides as the DAG count grows, and throws away the `@daily` intent |
| Hash the DAG id into a literal cron string | Same loss of intent; not reusable across DAGs/teams; every author re-implements it ad hoc |
| Pools / concurrency limits (`parallelism`, `max_active_tasks_per_dag`, pool slots) | Cap how many tasks run at once, but the runs are still *scheduled* at the same instant. They manage contention downstream; jitter reduces the peak at the source. The two are complementary, not alternatives. |
Jitter is peak shaving on top of concurrency control, not a replacement for it. It is the same idea as Jenkins' `H` cron syntax. The offset is deterministic: the same `seed` (e.g. the DAG id) always maps to the same offset, so runs stay stable across scheduler restarts and serialization. Motivated by the discussion in https://github.com/apache/airflow/discussions/69027.
### What
- `seed`/`max_jitter` added to `CronMixin` in both layers, so all four cron timetables inherit them; the concrete classes thread the params through their constructors and `serialize`/`deserialize`, and the `encoders.py` variants emit them.
- The offset is `md5(seed) % max_jitter`, computed in integer microseconds (FIPS-safe `hashlib_wrapper.md5`), and applied as a "strip → cron → apply" coordinate shift in `CronMixin._get_next`/`_get_prev`. Because `_align_to_next`/`_align_to_prev` build on those primitives, every cron timetable inherits the shift with no per-class scheduling code.
- **Fully opt-in and safe by default**: with `max_jitter` at its default of zero the offset is zero and every timetable behaves *identically* to before. Nothing changes for anyone who doesn't use it.
### Design decisions
- **Uniform shift for data-interval timetables.** The cron boundaries define the window, so the whole interval moves by the offset: same length, consecutive runs stay contiguous, but it no longer starts exactly on the cron time (00:35 to 00:35 instead of 00:00 to 00:00). Documented, with guidance to keep `max_jitter` small relative to the period.
- **`MultipleCronTriggerTimetable` children share one offset.** The same `seed`/`max_jitter` is passed to every child, so the whole DAG shifts in lockstep and its fire times keep their relative spacing.
- **Serialization only emits `seed`/`max_jitter` when jitter is set.** The wire format of existing DAGs is unchanged (so nothing is re-serialized on upgrade) and `deserialize` tolerates missing keys. The existing serialization tests pass untouched.
- **Jitter is part of `__eq__`/`__hash__`**, since timetables differing only in jitter produce different schedules. While adding this I fixed a latent bug: `__hash__` included the pendulum `Timezone` object, which is unhashable, so `hash()` on any cron timetable raised; it now hashes `str(timezone)`.
- **Empty `seed` with `max_jitter > 0` raises** in both layers: it would give every DAG the same offset and merely move the herd instead of spreading it.
### Tests
`airflow-core/tests/unit/timetables/test_cron_timetable_jitter.py` (28 tests), modeled on the existing `test_trigger_timetable.py`:
- jittered runs equal plain cron runs shifted by the fixed offset, across a catchup sequence including a DST spring-forward (`America/New_York`);
- zero `max_jitter` reproduces the plain timetable exactly (catchup on/off);
- offsets are deterministic for a given seed, bounded to `[0, max_jitter)`, spread across distinct seeds, and correct for sub-second windows;
- the empty-seed guard fires in both layers for all four timetables;
- core `serialize`/`deserialize` and SDK encode → decode round-trips preserve the offset for all four timetables;
- data-interval jitter shifts both bounds uniformly and keeps consecutive runs contiguous;
- `MultipleCronTriggerTimetable` children share a single offset;
- un-jittered timetables serialize without the jitter keys;
- jitter participates in equality, and equal timetables hash equal.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude (Claude Code), following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
related: #69705
related: https://github.com/apache/airflow/discussions/69027
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-5888153981
One option thing to consider here is the Jenkins approach of using `H`: So `H * * * *` would be once per hour, as a fixed time per dag, but not the top of the hour.
```groovy
cron('H 2 * * *') // Daily around 2 AM (H spreads load)
```
https://crongenerator.dev/cron-jenkins/
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-5778044516
cc: @uranusjr @kaxil -> I think that one should be reviewed by you :)
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-5888929157
Good pointer. croniter already supports `H` via `hash_id`, so this would be small to add and would replace the offset machinery entirely. The catch is the same as before: `H` needs a per Dag `hash_id`, so the dag_id or an explicit seed.
Happy to switch this PR to `H` syntax if that is preferred over `seed`/`max_jitter`, or to add it as a follow up. Which would you rather?
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-5875344384
Thanks for the thorough review @potiuk, all points addressed and rebased on main. `description` now shows the jitter offset (`summary` stays the bare cron), and the hash fix is split out as #73859.
On seed defaulting to the dag_id: agree the copied Dag footgun is real, I would do it as a follow up to keep this PR scoped. PTAL 🙏
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on code in PR #72475:
URL: https://github.com/apache/airflow/pull/72475#discussion_r4125246239
##########
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:
Fixed, keys now stay on the cron boundary while the run fires late. Added tests for both of your examples (including the coarse `key_format` date flip) and for backfill iteration with `run_offset` 0, 1 and -1. ✅
##########
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:
Fixed, now rejected in both layers. Added tests for core and SDK. ✅
##########
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:
Moved inside the jitter branch. ✅
##########
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:
Fixed on all lines this PR adds. ✅
##########
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:
Split out as #73859 with its own test. ✅
##########
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:
Added to `test__cron.py`: empty seed, negative window, and defaults being a no-op. ✅
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-6082573567
Thanks for catching this. Previous runs are now mapped to the tick they belong to before stepping, so enabling jitter or changing its settings no longer reruns the last tick. Added tests starting from an unjittered last run for all three timetable kinds, plus the run_offset coverage, the single description suffix, and seed ignored in eq/hash when jitter is off.
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-6096388553
I'll add H as a follow-up PR, with seed defaulting to the Dag id and used as the croniter hash_id, so this one keeps the scope agreed with potiuk.
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-6084414849
I think making `H` work hashed on dag_id would be a great addition to this, on top of any config based jitter.
One lets dag authors choose it, the other lets deployment admins force 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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-6085787488
Yeah sounds good will add that!
--
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] Add opt-in deterministic jitter to cron-based timetables [airflow]
Posted by "Pebble32 (via GitHub)" <gi...@apache.org>.
Pebble32 commented on PR #72475:
URL: https://github.com/apache/airflow/pull/72475#issuecomment-6082707717
Thanks for catching this. Previous runs are now mapped to the tick they belong to before stepping, so enabling jitter or changing its settings no longer reruns the last tick. Added tests starting from an unjittered last run for all three timetable kinds, plus a full off, on, off round trip checking one run per day and a return to the plain schedule. Also added the run_offset coverage, the single description suffix, and seed ignored in eq/hash when jitter is off
--
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