This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 101bc155e0f Thread version_data to callbacks (#69185)
101bc155e0f is described below
commit 101bc155e0fcec8b54ec0e21df0344aa2d2a5997
Author: Niko Oliveira <[email protected]>
AuthorDate: Tue Jul 7 15:35:31 2026 -0700
Thread version_data to callbacks (#69185)
PR #67217 wired version_data into the task execution path but not
callbacks, so callbacks for a pinned run initialized their bundle
without the manifest tasks used. Populate version_data on the
callback producer paths (ExecuteCallback.make and the
CallbackRequest creation sites) under the same pin guard as tasks,
carry it on BaseCallbackRequest, and forward it through
prepare_callback_bundle to get_bundle.
---
.../src/airflow/callbacks/callback_requests.py | 2 +
airflow-core/src/airflow/dag_processing/manager.py | 6 ++-
.../src/airflow/executors/workloads/callback.py | 4 ++
.../src/airflow/executors/workloads/task.py | 7 ++-
.../src/airflow/jobs/scheduler_job_runner.py | 10 +++-
airflow-core/src/airflow/models/dag_version.py | 14 ++++-
airflow-core/src/airflow/models/dagrun.py | 6 +++
airflow-core/src/airflow/models/trigger.py | 23 ++++++++-
.../tests/unit/callbacks/test_callback_requests.py | 27 ++++++++++
.../tests/unit/dag_processing/test_manager.py | 25 +++++++++
.../tests/unit/executors/test_workloads.py | 60 +++++++++++++++++++++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 1 +
airflow-core/tests/unit/models/test_dag_version.py | 29 +++++++++++
airflow-core/tests/unit/models/test_trigger.py | 27 ++++++++++
.../airflow/sdk/execution_time/schema/schema.json | 39 ++++++++++++++
ts-sdk/src/generated/supervisor.ts | 12 +++++
16 files changed, 282 insertions(+), 10 deletions(-)
diff --git a/airflow-core/src/airflow/callbacks/callback_requests.py
b/airflow-core/src/airflow/callbacks/callback_requests.py
index ce48438f0e7..6609dc30c6d 100644
--- a/airflow-core/src/airflow/callbacks/callback_requests.py
+++ b/airflow-core/src/airflow/callbacks/callback_requests.py
@@ -46,6 +46,8 @@ class BaseCallbackRequest(BaseModel):
"""File Path to use to run the callback"""
bundle_name: str
bundle_version: str | None
+ version_data: dict[str, Any] | None = None
+ """Optional structured metadata for the pinned bundle version (e.g. an S3
object manifest)."""
msg: str | None = None
"""Additional Message that can be used for logging to determine
failure/task heartbeat timeout"""
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index b2428985a65..f41883edf31 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -705,7 +705,11 @@ class DagFileProcessorManager(LoggingMixin):
Override to source the bundle from an API.
"""
try:
- bundle = DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
+ bundle = DagBundlesManager().get_bundle(
+ name=request.bundle_name,
+ version=request.bundle_version,
+ version_data=request.version_data,
+ )
except ValueError:
self.log.error("Bundle %s no longer configured, skipping
callback", request.bundle_name)
return None
diff --git a/airflow-core/src/airflow/executors/workloads/callback.py
b/airflow-core/src/airflow/executors/workloads/callback.py
index 04f26b8e787..c1842a59006 100644
--- a/airflow-core/src/airflow/executors/workloads/callback.py
+++ b/airflow-core/src/airflow/executors/workloads/callback.py
@@ -114,9 +114,13 @@ class ExecuteCallback(BaseDagBundleWorkload):
) -> ExecuteCallback:
"""Create an ExecuteCallback workload from a Callback ORM model."""
if not bundle_info:
+ from airflow.models.dag_version import _resolve_version_data
+
+ version_data = _resolve_version_data(dag_run.created_dag_version,
dag_run.bundle_version)
bundle_info = BundleInfo(
name=dag_run.dag_model.bundle_name,
version=dag_run.bundle_version,
+ version_data=version_data,
)
fname =
f"executor_callbacks/{dag_run.dag_id}/{dag_run.run_id}/{callback.id}"
diff --git a/airflow-core/src/airflow/executors/workloads/task.py
b/airflow-core/src/airflow/executors/workloads/task.py
index 611457f88a9..68a917118e3 100644
--- a/airflow-core/src/airflow/executors/workloads/task.py
+++ b/airflow-core/src/airflow/executors/workloads/task.py
@@ -102,13 +102,12 @@ class ExecuteTask(BaseDagBundleWorkload):
ser_ti = TaskInstanceDTO.model_validate(ti, from_attributes=True)
if not bundle_info:
- version_data = None
- if ti.dag_version is not None and ti.dag_run.bundle_version is not
None:
- version_data = ti.dag_version.version_data
+ from airflow.models.dag_version import _resolve_version_data
+
bundle_info = BundleInfo(
name=ti.dag_model.bundle_name,
version=ti.dag_run.bundle_version,
- version_data=version_data,
+ version_data=_resolve_version_data(ti.dag_version,
ti.dag_run.bundle_version),
)
fname = log_filename_template_renderer()(ti=ti)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index f4e66fc17c4..b941a5b3580 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -97,7 +97,7 @@ from airflow.models.connection_test import (
ConnectionTestState,
)
from airflow.models.dag import DagModel
-from airflow.models.dag_version import DagVersion
+from airflow.models.dag_version import DagVersion, _resolve_version_data
from airflow.models.dagbag import DBDagBag
from airflow.models.dagbundle import DagBundleModel
from airflow.models.dagrun import DagRun
@@ -1526,6 +1526,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if ti.dag_version and ti.dag_run.bundle_version is not
None
else ti.dag_run.bundle_version
)
+ _version_data = _resolve_version_data(ti.dag_version,
ti.dag_run.bundle_version)
# Backfill dag_version_id for legacy tasks (Pydantic
requires uuid.UUID).
if not _ensure_ti_has_dag_version_id(ti, session,
cls.logger()):
continue
@@ -1533,6 +1534,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
filepath=ti.dag_model.relative_fileloc or "",
bundle_name=_bundle_name,
bundle_version=_bundle_version,
+ version_data=_version_data,
ti=ti,
msg=msg,
task_callback_type=(
@@ -1575,6 +1577,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
_email_bundle_version = (
ti.dag_version.bundle_version if ti.dag_version else
ti.dag_run.bundle_version
)
+ _email_version_data =
_resolve_version_data(ti.dag_version, ti.dag_run.bundle_version)
# Backfill dag_version_id for legacy tasks (Pydantic
requires uuid.UUID).
if not _ensure_ti_has_dag_version_id(ti, session,
cls.logger()):
continue
@@ -1582,6 +1585,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
filepath=ti.dag_model.relative_fileloc or "",
bundle_name=_email_bundle_name,
bundle_version=_email_bundle_version,
+ version_data=_email_version_data,
ti=ti,
msg=msg,
email_type="retry" if ti.is_eligible_to_retry() else
"failure",
@@ -3096,6 +3100,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if ti.dag_version and ti.dag_run.bundle_version is not
None
else ti.dag_run.bundle_version
)
+ _stuck_version_data =
_resolve_version_data(ti.dag_version, ti.dag_run.bundle_version)
# Backfill dag_version_id for legacy tasks (Pydantic
requires uuid.UUID).
# Note: we cannot use `continue` here because this method
is not
# inside a loop. If backfilling fails we simply skip the
callback.
@@ -3104,6 +3109,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
filepath=ti.dag_model.relative_fileloc or "",
bundle_name=_stuck_bundle_name,
bundle_version=_stuck_bundle_version,
+ version_data=_stuck_version_data,
ti=ti,
msg=msg,
context_from_server=TIRunContext(
@@ -3571,6 +3577,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if ti.dag_version and ti.dag_run.bundle_version is not None
else ti.dag_run.bundle_version
)
+ _hb_version_data = _resolve_version_data(ti.dag_version,
ti.dag_run.bundle_version)
# Backfill dag_version_id for legacy tasks (Pydantic requires
uuid.UUID).
if not _ensure_ti_has_dag_version_id(ti, session, self.log):
continue
@@ -3578,6 +3585,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
filepath=ti.dag_model.relative_fileloc or "",
bundle_name=_hb_bundle_name,
bundle_version=_hb_bundle_version,
+ version_data=_hb_version_data,
ti=ti,
msg=str(task_instance_heartbeat_timeout_message_details),
context_from_server=TIRunContext(
diff --git a/airflow-core/src/airflow/models/dag_version.py
b/airflow-core/src/airflow/models/dag_version.py
index e6564a6da06..eb0a5204956 100644
--- a/airflow-core/src/airflow/models/dag_version.py
+++ b/airflow-core/src/airflow/models/dag_version.py
@@ -19,7 +19,7 @@ from __future__ import annotations
import logging
from datetime import datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
from uuid import UUID
import sqlalchemy as sa
@@ -235,3 +235,15 @@ class DagVersion(Base):
def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+
+
+def _resolve_version_data(
+ dag_version: DagVersion | None, bundle_version: str | None
+) -> dict[str, Any] | None:
+ """Return a bundle version's ``version_data`` manifest, but only for
pinned runs."""
+ # Expose version_data only when the run is pinned (bundle_version set) and
a DagVersion is
+ # present, so the bundle initializes against the exact version the run
used. Unpinned runs
+ # follow the latest bundle state, and legacy rows have no DagVersion.
+ if dag_version is not None and bundle_version is not None:
+ return dag_version.version_data
+ return None
diff --git a/airflow-core/src/airflow/models/dagrun.py
b/airflow-core/src/airflow/models/dagrun.py
index 056151cefa1..27751ed8b56 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -1449,12 +1449,18 @@ class DagRun(Base, LoggingMixin):
)
relevant_ti = None
if not execute:
+ from airflow.models.dag_version import _resolve_version_data
+
+ # Only carry version_data for pinned runs so the callback
initializes the bundle
+ # against the same version the run used.
+ version_data = _resolve_version_data(self.created_dag_version,
self.bundle_version)
return DagCallbackRequest(
filepath=self.dag_model.relative_fileloc,
dag_id=self.dag_id,
run_id=self.run_id,
bundle_name=self.dag_model.bundle_name,
bundle_version=self.bundle_version,
+ version_data=version_data,
context_from_server=DagRunContext(
dag_run=self,
last_ti=relevant_ti,
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index e457b2ec936..d52daa649bc 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -570,12 +570,31 @@ def _(event: BaseTaskEndEvent, *, task_instance:
TaskInstance, session: Session)
if event.task_instance_state in (TaskInstanceState.SUCCESS,
TaskInstanceState.FAILED):
if task_instance.dag_model.relative_fileloc is None:
raise RuntimeError("relative_fileloc should not be None for a
finished task")
+ from airflow.models.dag_version import _resolve_version_data
+
+ # Derive bundle identity from the TI's dag_version (falling back
to dag_run/dag_model
+ # for legacy/unpinned runs), mirroring the other callback sites so
bundle_version and
+ # version_data always describe the same version.
+ bundle_name = (
+ task_instance.dag_version.bundle_name
+ if task_instance.dag_version
+ else task_instance.dag_model.bundle_name
+ )
+ bundle_version = (
+ task_instance.dag_version.bundle_version
+ if task_instance.dag_version and
task_instance.dag_run.bundle_version is not None
+ else task_instance.dag_run.bundle_version
+ )
+ version_data = _resolve_version_data(
+ task_instance.dag_version, task_instance.dag_run.bundle_version
+ )
request = TaskCallbackRequest(
filepath=task_instance.dag_model.relative_fileloc,
ti=task_instance,
task_callback_type=event.task_instance_state,
- bundle_name=task_instance.dag_model.bundle_name,
- bundle_version=task_instance.dag_run.bundle_version,
+ bundle_name=bundle_name,
+ bundle_version=bundle_version,
+ version_data=version_data,
)
log.info("Sending callback: %s", request)
try:
diff --git a/airflow-core/tests/unit/callbacks/test_callback_requests.py
b/airflow-core/tests/unit/callbacks/test_callback_requests.py
index 434223e3747..9b85d61e5db 100644
--- a/airflow-core/tests/unit/callbacks/test_callback_requests.py
+++ b/airflow-core/tests/unit/callbacks/test_callback_requests.py
@@ -118,6 +118,33 @@ class TestCallbackRequest:
assert request.is_failure_callback == expected_is_failure
+ def test_version_data_round_trips_and_defaults_none(self):
+ """version_data survives JSON serialization and defaults to None when
omitted."""
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py":
"ver123"}}
+ request = DagCallbackRequest(
+ filepath="filepath",
+ dag_id="fake_dag",
+ run_id="fake_run",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version="abc123",
+ version_data=version_data,
+ )
+ result = DagCallbackRequest.from_json(request.to_json())
+ assert result.version_data == version_data
+
+ # Omitted -> defaults to None and round-trips as None.
+ unpinned = DagCallbackRequest(
+ filepath="filepath",
+ dag_id="fake_dag",
+ run_id="fake_run",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version=None,
+ )
+ assert unpinned.version_data is None
+ assert DagCallbackRequest.from_json(unpinned.to_json()).version_data
is None
+
class TestDagRunContext:
def test_dagrun_context_creation(self):
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 4b55b6b2f1e..ee2d7cd76d4 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -1375,6 +1375,7 @@ class TestDagFileProcessorManager:
"filepath": "dag_callback_dag.py",
"bundle_name": "testing",
"bundle_version": None,
+ "version_data": None,
"msg": None,
"dag_id": "dag_id",
"run_id": "run_id",
@@ -1966,6 +1967,30 @@ class TestDagFileProcessorManager:
assert manager.prepare_callback_bundle(request) is bundle
bundle.initialize.assert_called_once()
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager")
+ def test_prepare_callback_bundle_forwards_version_data(self,
mock_bundle_manager):
+ manager = DagFileProcessorManager(max_runs=1)
+ bundle = MagicMock(spec=BaseDagBundle)
+ bundle.supports_versioning = True
+ mock_bundle_manager.return_value.get_bundle.return_value = bundle
+
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py":
"ver123"}}
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version="some_commit_hash",
+ version_data=version_data,
+ msg=None,
+ )
+
+ manager.prepare_callback_bundle(request)
+ mock_bundle_manager.return_value.get_bundle.assert_called_once_with(
+ name="testing", version="some_commit_hash",
version_data=version_data
+ )
+
@mock.patch("airflow.dag_processing.manager.DagBundlesManager")
def
test_prepare_callback_bundle_skips_initialize_for_unversioned_request(self,
mock_bundle_manager):
manager = DagFileProcessorManager(max_runs=1)
diff --git a/airflow-core/tests/unit/executors/test_workloads.py
b/airflow-core/tests/unit/executors/test_workloads.py
index 63a249a36fd..1ef04477d08 100644
--- a/airflow-core/tests/unit/executors/test_workloads.py
+++ b/airflow-core/tests/unit/executors/test_workloads.py
@@ -28,7 +28,7 @@ from airflow.api_fastapi.auth.tokens import JWTGenerator
from airflow.executors import workloads
from airflow.executors.workloads import TaskInstance, TaskInstanceDTO, base as
workloads_base
from airflow.executors.workloads.base import BaseWorkloadSchema, BundleInfo
-from airflow.executors.workloads.callback import CallbackDTO,
CallbackFetchMethod
+from airflow.executors.workloads.callback import CallbackDTO,
CallbackFetchMethod, ExecuteCallback
from airflow.executors.workloads.task import ExecuteTask
from airflow.executors.workloads.types import state_class_for_key
from airflow.models.callback import CallbackKey
@@ -250,3 +250,61 @@ class TestExecuteTaskMakeVersionData:
assert workload.bundle_info.version == "abc123"
assert workload.bundle_info.version_data is None
+
+
+class TestExecuteCallbackMakeVersionData:
+ """Tests for ExecuteCallback.make() threading version_data through
BundleInfo."""
+
+ @staticmethod
+ def _make_mocks(bundle_version, version_data, *,
has_created_dag_version=True):
+ """Build mock Callback + DagRun with the attributes
ExecuteCallback.make() reads."""
+ from unittest.mock import Mock
+
+ callback = Mock()
+ callback.id = uuid4()
+ callback.fetch_method = CallbackFetchMethod.IMPORT_PATH
+ callback.data = {"path": "my_module.my_callback"}
+
+ dag_run = Mock()
+ dag_run.dag_id = "test_dag"
+ dag_run.run_id = "test_run"
+ dag_run.bundle_version = bundle_version
+ dag_run.dag_model.bundle_name = "test-bundle"
+ dag_run.dag_model.relative_fileloc = "dags/test_dag.py"
+ if has_created_dag_version:
+ dag_run.created_dag_version.version_data = version_data
+ else:
+ dag_run.created_dag_version = None
+
+ return callback, dag_run
+
+ def test_pinned_run_populates_version_data(self):
+ """When the run is pinned, version_data from created_dag_version flows
to BundleInfo."""
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py":
"ver123"}}
+ callback, dag_run = self._make_mocks(bundle_version="abc123",
version_data=version_data)
+
+ workload = ExecuteCallback.make(callback=callback, dag_run=dag_run)
+
+ assert workload.bundle_info.version == "abc123"
+ assert workload.bundle_info.version_data == version_data
+
+ def test_unpinned_run_suppresses_present_version_data(self):
+ """An unpinned run must not expose version_data even when
created_dag_version carries it."""
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py":
"ver123"}}
+ callback, dag_run = self._make_mocks(bundle_version=None,
version_data=version_data)
+
+ workload = ExecuteCallback.make(callback=callback, dag_run=dag_run)
+
+ assert workload.bundle_info.version is None
+ assert workload.bundle_info.version_data is None
+
+ def test_missing_created_dag_version_yields_none(self):
+ """A pinned run without a created_dag_version yields no
version_data."""
+ callback, dag_run = self._make_mocks(
+ bundle_version="abc123", version_data=None,
has_created_dag_version=False
+ )
+
+ workload = ExecuteCallback.make(callback=callback, dag_run=dag_run)
+
+ assert workload.bundle_info.version == "abc123"
+ assert workload.bundle_info.version_data is None
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index ef5ea043060..59bc50a7f12 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -730,6 +730,7 @@ class TestSchedulerJob:
ti=mock.ANY,
bundle_name="dag_maker",
bundle_version=None,
+ version_data=None,
msg=f"Executor {executor} reported that the task instance "
f"<TaskInstance:
test_process_executor_events_with_callback.dummy_task test [queued]
ti_id={ti1.id}> "
"finished with state failed, but the task instance's state
attribute is queued. "
diff --git a/airflow-core/tests/unit/models/test_dag_version.py
b/airflow-core/tests/unit/models/test_dag_version.py
index ec600c7be66..da4923d8771 100644
--- a/airflow-core/tests/unit/models/test_dag_version.py
+++ b/airflow-core/tests/unit/models/test_dag_version.py
@@ -17,6 +17,7 @@
from __future__ import annotations
from datetime import timedelta
+from unittest import mock
import pytest
from sqlalchemy import func, select
@@ -181,3 +182,31 @@ class TestDagVersion:
retrieved = DagVersion.get_latest_version("test_no_version_data",
session=session)
assert retrieved.version_data is None
assert retrieved.bundle_version == "abc123"
+
+
+class TestResolveVersionData:
+ """Unit tests for the _resolve_version_data pin-guard helper."""
+
+ @pytest.mark.parametrize(
+ ("dag_version", "bundle_version", "expected"),
+ [
+ pytest.param(
+ mock.Mock(version_data={"schema_version": 1}),
+ "abc123",
+ {"schema_version": 1},
+ id="pinned-with-data",
+ ),
+ pytest.param(
+ mock.Mock(version_data={"schema_version": 1}),
+ None,
+ None,
+ id="unpinned-suppresses-present-data",
+ ),
+ pytest.param(None, "abc123", None, id="missing-dag-version"),
+ pytest.param(None, None, None, id="unpinned-and-missing"),
+ ],
+ )
+ def test_resolve_version_data(self, dag_version, bundle_version, expected):
+ from airflow.models.dag_version import _resolve_version_data
+
+ assert _resolve_version_data(dag_version, bundle_version) == expected
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index 669d98a7704..91a6a92e27c 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -335,6 +335,33 @@ def test_submit_event_task_end(mock_utcnow, session,
create_task_instance, event
assert actual_xcoms == expected_xcoms
+@patch("airflow.callbacks.database_callback_sink.DatabaseCallbackSink.send")
+def test_submit_event_task_end_callback_includes_version_data(mock_send,
session, create_task_instance):
+ """A finished deferred task's callback carries version_data for a pinned
run, and its
+ bundle_version is derived from the same dag_version so the two cannot
diverge."""
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py": "ver123"}}
+
+ trigger = Trigger(classpath="does.not.matter", kwargs={})
+ session.add(trigger)
+ task_instance = create_task_instance(
+ session=session, logical_date=timezone.utcnow(), state=State.DEFERRED
+ )
+ task_instance.trigger_id = trigger.id
+ # Pin the run and attach a manifest to the TI's dag_version.
+ task_instance.dag_run.bundle_version = "some_hash"
+ task_instance.dag_version.bundle_version = "some_hash"
+ task_instance.dag_version.version_data = version_data
+ session.commit()
+
+ Trigger.submit_event(trigger.id, TaskSuccessEvent(), session=session)
+ session.flush()
+
+ mock_send.assert_called_once()
+ request = mock_send.call_args.kwargs["callback"]
+ assert request.bundle_version == "some_hash"
+ assert request.version_data == version_data
+
+
@pytest.fixture
def create_triggerer():
"""Fixture factory which creates individual test Triggerer instances."""
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index 643bda0a5c6..01e3c8970b5 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -753,6 +753,19 @@
],
"title": "Bundle Version"
},
+ "version_data": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Version Data"
+ },
"msg": {
"anyOf": [
{
@@ -1651,6 +1664,19 @@
],
"title": "Bundle Version"
},
+ "version_data": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Version Data"
+ },
"msg": {
"anyOf": [
{
@@ -3889,6 +3915,19 @@
],
"title": "Bundle Version"
},
+ "version_data": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Version Data"
+ },
"msg": {
"anyOf": [
{
diff --git a/ts-sdk/src/generated/supervisor.ts
b/ts-sdk/src/generated/supervisor.ts
index fe61155f39e..004ad49e7ba 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -120,6 +120,9 @@ export type Type11 = "DRCount";
export type Filepath = string;
export type BundleName = string;
export type BundleVersion = string | null;
+export type VersionData1 = {
+ [k: string]: unknown;
+} | null;
export type Msg = string | null;
export type DagId1 = string;
export type RunId1 = string;
@@ -192,6 +195,9 @@ export type BundleName1 = string;
export type Filepath1 = string;
export type BundleName2 = string;
export type BundleVersion1 = string | null;
+export type VersionData2 = {
+ [k: string]: unknown;
+} | null;
export type Msg1 = string | null;
/**
* All possible states that a Task Instance can be in.
@@ -243,6 +249,9 @@ export type Type13 = "TaskCallbackRequest";
export type Filepath2 = string;
export type BundleName3 = string;
export type BundleVersion2 = string | null;
+export type VersionData3 = {
+ [k: string]: unknown;
+} | null;
export type Msg2 = string | null;
export type EmailType = "failure" | "retry";
export type Type14 = "EmailRequest";
@@ -849,6 +858,7 @@ export interface DagCallbackRequest {
filepath: Filepath;
bundle_name: BundleName;
bundle_version: BundleVersion;
+ version_data?: VersionData1;
msg?: Msg;
dag_id: DagId1;
run_id: RunId1;
@@ -959,6 +969,7 @@ export interface TaskCallbackRequest {
filepath: Filepath1;
bundle_name: BundleName2;
bundle_version: BundleVersion1;
+ version_data?: VersionData2;
msg?: Msg1;
ti: TaskInstance;
task_callback_type?: TaskInstanceState | null;
@@ -1019,6 +1030,7 @@ export interface EmailRequest {
filepath: Filepath2;
bundle_name: BundleName3;
bundle_version: BundleVersion2;
+ version_data?: VersionData3;
msg?: Msg2;
ti: TaskInstance;
email_type?: EmailType;
