Airflow 3.3.0 shipped on 6 July 2026, with 3.3.1 following about five weeks later. It's a much smaller release than 3.0 was, and that's fine. Nothing about how you write Dags has changed.
What did change is a set of things most of us have been working around for years. Tasks that need to remember a cursor across a retry. Daily jobs that fire before all their hourly inputs land. Retry loops that keep hammering an endpoint returning 401. Reruns that quietly pick up code from three deploys ago. Approval steps that burn triggerer capacity while a human takes two days to respond.
This post walks through what's in 3.3, what problem each piece solves, and what it looks like in an actual pipeline.
What Airflow is, briefly
Airflow is an open-source orchestrator. You describe workflows as Dags in Python, and the scheduler works out when each task runs, where, in what order, and what happens when something breaks.
It doesn't move data itself. It calls the systems that do: warehouses, Spark, dbt, object storage, whatever API you're stuck integrating with this quarter. The value is in dependency resolution, retries, backfills, and a durable record of what ran.
Since 3.0 the architecture has been service-oriented. Tasks talk to a Task Execution API rather than reaching into the metadata database, and authoring happens against a stable airflow.sdk interface. 3.3 builds on that; it doesn't rearrange it.
Why this matters for data engineering work
A pipeline is almost never one job. It's ingestion, validation, transformation, publication, plus a set of downstream consumers who care a great deal about when the output is ready and usually work on a different team.
Airflow is where those pieces get joined up. Dependencies get expressed once instead of encoded as cron offsets that someone tuned in 2023 and nobody dares touch. Failures have defined semantics. And when something goes wrong at 3am, there's a run history to argue from.
Most of 3.3 improves the second and third of those.
The short version of what's new
Five changes carry the weight:
- a first-class state store for tasks and assets (AIP-103)
- task implementations in Java and Go via a new Coordinator layer (AIP-108, experimental)
- a big expansion of asset partitioning, which first appeared in 3.2
- pluggable retry policies (AIP-105)
- per-run control over which Dag bundle version a clear, rerun, or backfill executes
Plus a new scheduler-managed state for human-in-the-loop tasks, and two OpenTelemetry metric changes that will break your dashboards if you don't plan for them.
Task and asset state store
Tasks can now persist arbitrary key-value state that survives retries and reruns, through a task_state_store accessor on the Task SDK. Assets get their own asset_state_store, scoped to the asset rather than the task instance, so every task that produces or consumes that asset reads and writes the same keys.
Before this, "remember where I stopped" meant one of three things: an XCom (scoped to the run, so useless across runs), a Variable abused as a mutable scratchpad (global, unversioned, and a lovely source of race conditions), or a side table in your own database (one more thing to migrate, back up, and clean up). Every team I've worked with has picked one of the three and regretted it later.
State lives in the metadata database by default, with per-key retention and periodic garbage collection. There's a clear_on_success option if you want task state wiped after a clean run, and max_value_storage_bytes caps REST API writes at 64 KB out of the box. If you need bigger payloads or credentialed storage, point [workers] state_store_backend at your own worker-side backend and the Execution API stores a reference string instead. Both stores are readable through the Core and Execution APIs, and both have UI views, including a column showing which task instance last wrote each entry.
The concrete case: an incremental extract reads last_cursor, pulls records after it, writes the new high-water mark before exiting. Fail halfway, retry, resume from the cursor instead of re-reading the whole window. If several Dags share a watermark on a warehouse table, that goes in the asset state store instead so producer and consumers all see the same value.
The most convincing demonstration is ResumableJobMixin, which SparkSubmitOperator now uses. With durable=True (the default), a worker that dies mid-run reconnects to the existing Spark job on retry rather than submitting a second one. Anyone who has paid for two identical Spark jobs because a spot instance got reclaimed will appreciate that.
Duplicate submissions and full re-reads are the two genuinely expensive failure modes in long-running ingestion. Making state a platform concern rather than a per-team hack is what makes both avoidable.
Asset partitioning
Asset events can carry a partition_key, which lets you model the same asset at partition granularity: an hour, a day, a region, a tenant. That arrived in 3.2. What 3.3 adds is much better control over how one upstream event maps to downstream runs.
The new pieces are FanOutMapper for one-to-many fan-out, FixedKeyMapper paired with SegmentWindow for categorical rollups, wait policies (WaitForAll or MinimumCount(n)), windows that enumerate forward or backward from their anchor, a per-mapper max_downstream_keys cap alongside the global [scheduler] partition_mapper_max_downstream_keys bound (default 1000), and a PartitionedAtRuntime timetable for when the key is only known once the producing task runs.
The problem this fixes is that cross-Dag dependencies used to be all or nothing. A daily summary scheduled on an hourly asset fires on the first event of the day, which is wrong, so you build sensors, or you offset the cron by six hours and hope ingestion never runs late. Neither actually says "wait until all 24 of today's hourly partitions have landed."
RollupMapper says it. It composes an upstream mapper with a window, and the scheduler holds the downstream run until the window is satisfied. Partial windows show up as pending on the next-run-assets view, so when the daily job hasn't started you can see which hour is missing rather than reverse-engineering it from logs. MinimumCount(n) covers the case where you'd rather fire on 22 of 24 hours than block the whole day on one flaky source.
Categorical rollups work the same way. SegmentWindow(["us", "eu", "apac"]) with FixedKeyMapper("all_regions") holds the aggregate until all three regions arrive, then fires once with dag_run.partition_key set to all_regions.
Two things worth knowing before you build on this. Misconfigurations raise TypeError at Dag parse rather than silently holding runs forever, which is a good design decision. And DayWindow always enumerates 24 hourly steps, so a rollup configured in a DST-observing timezone will wait forever on the spring-forward day, when only 23 hours exist. Use UTC for anything crossing a DST boundary.
Java and Go tasks
Individual task implementations can now be written in Java or Go while the Dag and its scheduling stay in Python. You declare the task shape with @task.stub(queue=...), and the worker routes it to a configured coordinator: JavaCoordinator for the JVM, ExecutableCoordinator for self-contained native binaries like Go. The coordinator runs your code and proxies Variables, Connections, and XComs back through the Execution API.
Plenty of validation and transformation logic already exists in Java or Go, owned by people who don't write Python and have no interest in starting. Until now the options were a rewrite or a BashOperator wrapper that threw away logs, XComs, and retry semantics.
The integration is better than I expected. Native logs stream into the Airflow UI and honour your remote logging config. XComs round-trip in both directions, so a Python extract can hand off to a Go transform and then to a Java load. retries set on the Python stub is honoured by both SDKs. It runs on LocalExecutor, CeleryExecutor, and KubernetesExecutor, though the last needs extra configuration.
Now the caveat, and it's a real one: this is experimental in 3.3.0. The SDK APIs and wire protocol may change between releases. The Java and Go SDKs ship as separate artifacts (a Maven/Gradle dependency, and a Go module plus coordinator), not in the apache-airflow pip package. And the Dag shape is still declared in Python, so you're maintaining two files per cross-language task. Pilot it on something you can afford to rewrite. Don't standardise on it yet.
Pluggable retry policies
Retry behaviour is a parameter now, not a fixed count. An ExceptionRetryPolicy maps exception types to actions: RETRY with an optional custom delay, FAIL immediately, or DEFAULT to fall through to normal behaviour. Rules evaluate in order and the first match wins.
retries=3 treats a rate limit and an expired credential exactly the same. One is worth waiting five minutes for. The other is going to fail three more times over the next twelve minutes and then page someone, having learned nothing along the way.
A few details that matter when you build these:
- The policy runs in the task worker process, never the scheduler, and every decision is logged as
Retry policy decision action=... reason=.... You can see what it decided and why. -
retriesstays the ceiling. A policy can fail early but can't extend past it. - Matching uses
isinstanceby default, so a rule forOSErroralso catchesConnectionError. Passmatch_subclasses=Falseif you want exact types. - Exceptions can be classes or dotted import strings, validated at Dag parse time.
-
AirflowFailExceptionalways wins; the policy is never consulted for it.
For anything the declarative form can't express (HTTP status codes, Retry-After headers, response bodies) you subclass RetryPolicy and implement evaluate(), which receives the exception, try number, max tries, and the full context. Policies also work with dynamic task mapping through .partial(), shared across all mapped instances but evaluated per instance.
Most teams can name their non-retryable errors off the top of their head. Write two or three shared policies, apply them via default_args, and stop tuning retries task by task.
Dag versioning
Dag versioning itself landed in 3.0 under AIP-66. What 3.3 adds is control over which bundle version a cleared, rerun, or backfilled run actually executes, through rerun_with_latest_version.
Airflow 2.x always reran with the latest code. Airflow 3.x defaults to the original version from the initial run. Both defaults are wrong some of the time. Reproducing an incident wants the original code. Rerunning after a bug fix obviously wants the fix.
Resolution goes by precedence: an explicit request parameter or CLI flag (run_on_latest_version, exposed as a CLI flag for backfill only), then the Dag-level rerun_with_latest_version, then [core] rerun_with_latest_version, then the historical defaults, which are False for clear and rerun and True for backfills. It applies to TriggerDagRunOperator reruns too.
So: transformation had an off-by-one in its date filter, you ship the fix, you backfill three weeks on the latest bundle. Later, an auditor asks why last month's numbers looked odd, and you clear those runs on the original version so the behaviour matches what actually happened. Both are one flag.
Set [core] rerun_with_latest_version deliberately and write it down somewhere, or your team will discover the default during an incident, which is the worst time to learn it.
Observability
Three changes, and two of them will break things.
OpenTelemetry timer and timing metrics are now recorded as Histograms instead of Gauges, preserving count, sum, and bucket distribution across recordings. Gauge-recorded timers only ever gave you the last value, which makes latency percentiles impossible. Now "p95 task duration for this Dag over the last hour" is a question with an answer.
The Dag-processing metric dag_processing.last_run.seconds_ago is now emitted with file_path, bundle_name, and file_name tags instead of baking the filename into the metric path. The legacy form is still emitted by default and can be turned off with [metrics] legacy_names_on.
Both of those are breaking for existing dashboards and alerts. Anything built on the old Gauge series, or parsing the old ...seconds_ago.{dag_file} path, needs rewriting. Inventory your alerts before the upgrade, not after the first missed page.
The third change is additive: a Deadlines page under the Browse menu, visible to any role that already has can_read and menu_access on Dag Runs. 3.3 also adds OpenTelemetry head sampling and propagates trace context from the client into Execution API server-side spans.
Human-in-the-loop workflows
The HITL operators arrived in 3.1: HITLOperator for option selection, HITLEntryOperator for free-form input, ApprovalOperator for approve/reject, HITLBranchOperator for branch selection. 3.3 changes how a waiting task is held, and the change is bigger than it sounds.
A task awaiting a human response now sits in a dedicated, scheduler-managed awaiting_input state instead of deferring onto the triggerer. While it waits it holds no worker slot, no triggerer, and no pool slot. That last one is a real difference from the old path, where a deferred HITL task counted against any pool with include_deferred enabled. The triggerer can now scale to zero even with tasks parked waiting for input. Tasks resume on a response or on the scheduler's response-timeout sweep. On 3.1 and 3.2, HITL tasks still use trigger-based deferral.
Approval steps wait hours or days. Under the old model that idle time consumed capacity, which turned "add a review gate" into a capacity planning conversation and, in practice, into a shared spreadsheet instead.
The surrounding tooling is decent. assigned_users restricts who may respond. response_timeout with a defaults value covers the case where nobody does. Notifiers can send an actionable link, built with HITLOperator.generate_link_to_ui_from_context. Responses can also come through the REST API (PATCH .../hitlDetails), and airflow dags test now waits properly for input rather than looping on parked tasks.
A typical shape: a curated pricing table rebuilt nightly, with an ApprovalOperator in front of publication showing the row-count delta and a few sanity checks. Steward approves and the publish task runs. Steward rejects, or the five-hour timeout expires with defaults="Reject", and publication is skipped. The decision ends up recorded next to the run it governed instead of in a Slack thread nobody can find in March.
Putting it together
Say you run a marketing analytics platform. Clickstream events land hourly in object storage, get validated and normalised, roll up into a daily curated table, get reviewed when the volume delta looks strange, and only then reach the BI layer.
The consumer Dag:
from datetime import timedelta
from airflow.sdk import (
DAG,
Asset,
DayWindow,
ExceptionRetryPolicy,
MinimumCount,
PartitionedAssetTimetable,
RetryAction,
RetryRule,
RollupMapper,
StartOfHourMapper,
task,
)
hourly_events = Asset(uri="s3://raw/events/hourly", name="hourly_events")
WAREHOUSE_RETRY_POLICY = ExceptionRetryPolicy(
rules=[
RetryRule(
exception="requests.exceptions.HTTPError",
action=RetryAction.RETRY,
retry_delay=timedelta(minutes=5),
reason="Warehouse API rate limit",
),
RetryRule(
exception="google.auth.exceptions.RefreshError",
action=RetryAction.FAIL,
reason="Credential failure is not retryable",
),
],
)
with DAG(
dag_id="daily_curated_events",
schedule=PartitionedAssetTimetable(
assets=hourly_events,
default_partition_mapper=RollupMapper(
upstream_mapper=StartOfHourMapper(),
window=DayWindow(),
# Fire once 22 of the day's 24 hourly partitions are in.
wait_policy=MinimumCount(22),
),
),
catchup=False,
):
@task(retries=3, retry_policy=WAREHOUSE_RETRY_POLICY)
def load_partition(**context):
task_state = context["task_state_store"]
day = context["dag_run"].partition_key # e.g. "2026-09-02"
cursor = task_state.get("last_cursor", default=0)
new_cursor = load_rows_for(day, since=cursor) # your loader
task_state.set("last_cursor", new_cursor)
load_partition()
Four things are earning their keep here. The rollup mapper stops the daily run starting on partial data, while MinimumCount(22) keeps one dead hour from blocking the whole day. The retry policy tells a rate limit apart from a credential failure. The state store makes a half-finished load resumable instead of restartable. And the approval gate downstream costs nothing while it waits.
What to actually do about it
Upgrade with the metrics work scoped in. The Histogram switch and the tagged Dag-processing metric are the two changes most likely to break quietly, and rewriting affected queries takes longer than the upgrade itself. Read the Significant Changes section of the release notes first.
Move your cursors and watermarks out of Variables. Start with pipelines where a retry currently means reprocessing a full window, since that's where the payback is immediate.
Write two or three retry policies rather than tuning retries everywhere.
If you're currently using sensors or cron offsets to say "wait for all of yesterday's inputs," look at partitioned assets. They say it directly, and they tell you which partition is missing when the run doesn't fire.
Try the Language Task SDK on something non-critical if you have Java or Go logic sitting outside the platform. Keep it out of your core pipeline until it stops being experimental.
Wrapping up
3.3 doesn't change how you write Dags. It reduces how much you have to work around Airflow when a pipeline meets reality: a task that needs memory, a job that needs complete inputs, an error that shouldn't be retried, a rerun that needs a specific version of the code, a step that needs a person.
If you only look at two things, make them the state store and retry policies. Both replace patterns nearly every team has hand-rolled badly at least once. Asset partitioning is the bigger structural win if your dependency graph spans Dags. The multi-language SDK is worth a look and not yet worth a bet.
The 3.3.0 release notes are the thing to read before you upgrade.
Top comments (0)