Merge branch 'main' into dag_run_migrate_conf_column_as_json

.github/ISSUE_TEMPLATE/airflow_bug_report.yml

@@ -25,7 +25,7 @@ body:
         the latest release or main to see if the issue is fixed before reporting it.
       multiple: false
       options:
-        - "2.10.3"
+        - "2.10.4"
         - "main (development)"
         - "Other Airflow 2 version (please specify below)"
     validations:

.github/workflows/ci.yml

@@ -744,23 +744,17 @@ jobs:
     steps:
       - name: Notify Slack
         id: slack
-        uses: slackapi/slack-github-action@v1.27.0
+        uses: slackapi/slack-github-action@485a9d42d3a73031f12ec201c457e2162c45d02d  # v2.0.0
         with:
-          channel-id: 'internal-airflow-ci-cd'
+          method: chat.postMessage
+          token: ${{ env.SLACK_BOT_TOKEN }}
           # yamllint disable rule:line-length
           payload: |
-            {
-              "text": "🚨🕒 Scheduled CI Failure Alert 🕒🚨\n\n*Details:* <https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}|View the failure log>",
-              "blocks": [
-                {
-                  "type": "section",
-                  "text": {
-                    "type": "mrkdwn",
-                    "text": "🚨🕒 Scheduled CI Failure Alert 🕒🚨\n\n*Details:* <https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}|View the failure log>"
-                  }
-                }
-              ]
-            }
+            channel: "internal-airflow-ci-cd"
+            text: "🚨🕒 Scheduled CI Failure Alert 🕒🚨\n\n*Details:* <https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}|View the failure log>"
+            blocks:
+              - type: "section"
+                text:
+                  type: "mrkdwn"
+                  text: "🚨🕒 Scheduled CI Failure Alert 🕒🚨\n\n*Details:* <https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}|View the failure log>"
           # yamllint enable rule:line-length
-        env:
-          SLACK_BOT_TOKEN: ${{ env.SLACK_BOT_TOKEN }}

.pre-commit-config.yaml

@@ -491,21 +491,21 @@ repos:
         files: ^docs/apache-airflow/extra-packages-ref\.rst$|^hatch_build.py
         pass_filenames: false
         entry: ./scripts/ci/pre_commit/check_extra_packages_ref.py
-        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.26.3', 'tabulate']
+        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.27.0', 'tabulate']
       - id: check-hatch-build-order
         name: Check order of dependencies in hatch_build.py
         language: python
         files: ^hatch_build.py$
         pass_filenames: false
         entry: ./scripts/ci/pre_commit/check_order_hatch_build.py
-        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.26.3']
+        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.27.0']
       - id: update-extras
         name: Update extras in documentation
         entry: ./scripts/ci/pre_commit/insert_extras.py
         language: python
         files: ^contributing-docs/12_airflow_dependencies_and_extras.rst$|^INSTALL$|^providers/src/airflow/providers/.*/provider\.yaml$|^Dockerfile.*
         pass_filenames: false
-        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.26.3']
+        additional_dependencies: ['rich>=12.4.4', 'hatchling==1.27.0']
       - id: check-extras-order
         name: Check order of extras in Dockerfile
         entry: ./scripts/ci/pre_commit/check_order_dockerfile_extras.py

Dockerfile

@@ -45,7 +45,7 @@ ARG AIRFLOW_UID="50000"
 ARG AIRFLOW_USER_HOME_DIR=/home/airflow
 
 # latest released version here
-ARG AIRFLOW_VERSION="2.10.3"
+ARG AIRFLOW_VERSION="2.10.4"
 
 ARG PYTHON_BASE_IMAGE="python:3.9-slim-bookworm"
 
@@ -55,7 +55,7 @@ ARG PYTHON_BASE_IMAGE="python:3.9-slim-bookworm"
 # Also use `force pip` label on your PR to swap all places we use `uv` to `pip`
 ARG AIRFLOW_PIP_VERSION=24.3.1
 # ARG AIRFLOW_PIP_VERSION="git+https://github.com/pypa/pip.git@main"
-ARG AIRFLOW_UV_VERSION=0.5.8
+ARG AIRFLOW_UV_VERSION=0.5.9
 ARG AIRFLOW_USE_UV="false"
 ARG UV_HTTP_TIMEOUT="300"
 ARG AIRFLOW_IMAGE_REPOSITORY="https://github.com/apache/airflow"

Dockerfile.ci

@@ -1381,7 +1381,7 @@ RUN bash /scripts/docker/install_packaging_tools.sh; \
 # Also use `force pip` label on your PR to swap all places we use `uv` to `pip`
 ARG AIRFLOW_PIP_VERSION=24.3.1
 # ARG AIRFLOW_PIP_VERSION="git+https://github.com/pypa/pip.git@main"
-ARG AIRFLOW_UV_VERSION=0.5.8
+ARG AIRFLOW_UV_VERSION=0.5.9
 # TODO(potiuk): automate with upgrade check (possibly)
 ARG AIRFLOW_PRE_COMMIT_VERSION="4.0.1"
 ARG AIRFLOW_PRE_COMMIT_UV_VERSION="4.1.4"

PROVIDERS.rst

@@ -153,8 +153,8 @@ Airflow version to the next MINOR release, when 12 months passed since the first
 MINOR version of Airflow.
 
 For example this means that by default we upgrade the minimum version of Airflow supported by providers
-to 2.8.0 in the first Provider's release after 18th of December 2024. 18th of December 2023 is the date when the
-first ``PATCHLEVEL`` of 2.8 (2.8.0) has been released.
+to 2.9.0 in the first Provider's release after 8th of April 2025. 8th of April 2024 is the date when the
+first ``PATCHLEVEL`` of 2.9 (2.9.0) has been released.
 
 When we increase the minimum Airflow version, this is not a reason to bump ``MAJOR`` version of the providers
 (unless there are other breaking changes in the provider). The reason for that is that people who use

README.md

@@ -97,7 +97,7 @@ Airflow is not a streaming solution, but it is often used to process real-time d
 
 Apache Airflow is tested with:
 
-|            | Main version (dev)         | Stable version (2.10.3)    |
+|            | Main version (dev)         | Stable version (2.10.4)    |
 |------------|----------------------------|----------------------------|
 | Python     | 3.9, 3.10, 3.11, 3.12      | 3.8, 3.9, 3.10, 3.11, 3.12 |
 | Platform   | AMD64/ARM64(\*)            | AMD64/ARM64(\*)            |
@@ -175,15 +175,15 @@ them to the appropriate format and workflow that your tool requires.
 
 
 ```bash
-pip install 'apache-airflow==2.10.3' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.3/constraints-3.9.txt"
+pip install 'apache-airflow==2.10.4' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.9.txt"
 ```
 
 2. Installing with extras (i.e., postgres, google)
 
 ```bash
-pip install 'apache-airflow[postgres,google]==2.10.3' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.3/constraints-3.9.txt"
+pip install 'apache-airflow[postgres,google]==2.10.4' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.9.txt"
 ```
 
 For information on installing provider packages, check
@@ -288,7 +288,7 @@ Apache Airflow version life cycle:
 
 | Version   | Current Patch/Minor   | State     | First Release   | Limited Support   | EOL/Terminated   |
 |-----------|-----------------------|-----------|-----------------|-------------------|------------------|
-| 2         | 2.10.3                | Supported | Dec 17, 2020    | TBD               | TBD              |
+| 2         | 2.10.4                | Supported | Dec 17, 2020    | TBD               | TBD              |
 | 1.10      | 1.10.15               | EOL       | Aug 27, 2018    | Dec 17, 2020      | June 17, 2021    |
 | 1.9       | 1.9.0                 | EOL       | Jan 03, 2018    | Aug 27, 2018      | Aug 27, 2018     |
 | 1.8       | 1.8.2                 | EOL       | Mar 19, 2017    | Jan 03, 2018      | Jan 03, 2018     |

RELEASE_NOTES.rst

@@ -21,6 +21,52 @@
 
 .. towncrier release notes start
 
+Airflow 2.10.4 (2024-12-16)
+---------------------------
+
+Significant Changes
+^^^^^^^^^^^^^^^^^^^
+
+TaskInstance ``priority_weight`` is capped in 32-bit signed integer ranges (#43611)
+"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
+
+Some database engines are limited to 32-bit integer values. As some users reported errors in
+weight rolled-over to negative values, we decided to cap the value to the 32-bit integer. Even
+if internally in python smaller or larger values to 64 bit are supported, ``priority_weight`` is
+capped and only storing values from -2147483648 to 2147483647.
+
+Bug Fixes
+^^^^^^^^^
+
+- Fix stats of dynamic mapped tasks after automatic retries of failed tasks (#44300)
+- Fix wrong display of multi-line messages in the log after filtering (#44457)
+- Allow "/" in metrics validator (#42934) (#44515)
+- Fix gantt flickering (#44488) (#44517)
+- Fix problem with inability to remove fields from Connection form (#40421) (#44442)
+- Check pool_slots on partial task import instead of execution (#39724) (#42693)
+- Avoid grouping task instance stats by try_number for dynamic mapped tasks (#44300) (#44319)
+- Re-queue task when they are stuck in queued (#43520) (#44158)
+- Suppress the warnings where we check for sensitive values (#44148) (#44167)
+- Fix get_task_instance_try_details to return appropriate schema (#43830) (#44133)
+- Log message source details are grouped (#43681) (#44070)
+- Fix duplication of Task tries in the UI (#43891) (#43950)
+- Add correct mime-type in OpenAPI spec (#43879) (#43901)
+- Disable extra links button if link is null or empty (#43844) (#43851)
+- Disable XCom list ordering by execution_date (#43680) (#43696)
+- Fix venv numpy example which needs to be 1.26 at least to be working in Python 3.12 (#43659)
+- Fix Try Selector in Mapped Tasks also on Index 0 (#43590) (#43591)
+- Prevent using ``trigger_rule="always"`` in a dynamic mapped task (#43810)
+- Prevent using ``trigger_rule=TriggerRule.ALWAYS`` in a task-generated mapping within bare tasks (#44751)
+
+Doc Only Changes
+""""""""""""""""
+- Update XCom docs around containers/helm (#44570) (#44573)
+
+Miscellaneous
+"""""""""""""
+- Raise deprecation warning when accessing inlet or outlet events through str (#43922)
+
+
 Airflow 2.10.3 (2024-11-05)
 ---------------------------
 

airflow/api_fastapi/common/exceptions.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from enum import Enum
 from typing import Generic, TypeVar
 
 from fastapi import HTTPException, Request, status
@@ -38,26 +39,46 @@ def exception_handler(self, request: Request, exc: T):
         raise NotImplementedError
 
 
+class _DatabaseDialect(Enum):
+    SQLITE = "sqlite"
+    MYSQL = "mysql"
+    POSTGRES = "postgres"
+
+
 class _UniqueConstraintErrorHandler(BaseErrorHandler[IntegrityError]):
     """Exception raised when trying to insert a duplicate value in a unique column."""
 
+    unique_constraint_error_prefix_dict: dict[_DatabaseDialect, str] = {
+        _DatabaseDialect.SQLITE: "UNIQUE constraint failed",
+        _DatabaseDialect.MYSQL: "Duplicate entry",
+        _DatabaseDialect.POSTGRES: "violates unique constraint",
+    }
+
     def __init__(self):
         super().__init__(IntegrityError)
-        self.unique_constraint_error_messages = [
-            "UNIQUE constraint failed",  # SQLite
-            "Duplicate entry",  # MySQL
-            "violates unique constraint",  # PostgreSQL
-        ]
+        self.dialect: _DatabaseDialect.value | None = None
 
     def exception_handler(self, request: Request, exc: IntegrityError):
         """Handle IntegrityError exception."""
-        exc_orig_str = str(exc.orig)
-        if any(error_msg in exc_orig_str for error_msg in self.unique_constraint_error_messages):
+        if self._is_dialect_matched(exc):
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
-                detail="Unique constraint violation",
+                detail={
+                    "reason": "Unique constraint violation",
+                    "statement": str(exc.statement),
+                    "orig_error": str(exc.orig),
+                },
             )
 
+    def _is_dialect_matched(self, exc: IntegrityError) -> bool:
+        """Check if the exception matches the unique constraint error message for any dialect."""
+        exc_orig_str = str(exc.orig)
+        for dialect, error_msg in self.unique_constraint_error_prefix_dict.items():
+            if error_msg in exc_orig_str:
+                self.dialect = dialect
+                return True
+        return False
+
 
 DatabaseErrorHandlers = [
     _UniqueConstraintErrorHandler(),

airflow/api_fastapi/common/parameters.py

@@ -536,16 +536,17 @@ def _transform_dag_run_states(states: Iterable[str] | None) -> list[DagRunState
 
 # TI
 def _transform_ti_states(states: list[str] | None) -> list[TaskInstanceState | None] | None:
+    """Transform a list of state strings into a list of TaskInstanceState enums handling special 'None' cases."""
+    if not states:
+        return None
+
     try:
-        if not states:
-            return None
         return [None if s in ("none", None) else TaskInstanceState(s) for s in states]
     except ValueError:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail=f"Invalid value for state. Valid values are {', '.join(TaskInstanceState)}",
         )
-    return states
 
 
 QueryTIStateFilter = Annotated[

airflow/api_fastapi/core_api/datamodels/ui/structure.py

@@ -28,6 +28,7 @@ class EdgeResponse(BaseModel):
     label: str | None = None
     source_id: str
     target_id: str
+    is_source_asset: bool | None = None
 
 
 class NodeResponse(BaseModel):
@@ -41,6 +42,7 @@ class NodeResponse(BaseModel):
     setup_teardown_type: Literal["setup", "teardown"] | None = None
     type: Literal["join", "task", "asset-condition", "asset", "asset-alias", "dag", "sensor", "trigger"]
     operator: str | None = None
+    asset_condition_type: Literal["or-gate", "and-gate"] | None = None
 
 
 class StructureDataResponse(BaseModel):

airflow/api_fastapi/core_api/datamodels/xcom.py

@@ -33,6 +33,7 @@ class XComResponse(BaseModel):
     map_index: int
     task_id: str
     dag_id: str
+    run_id: str
 
 
 class XComResponseNative(XComResponse):

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -7734,6 +7734,11 @@ components:
         target_id:
           type: string
           title: Target Id
+        is_source_asset:
+          anyOf:
+          - type: boolean
+          - type: 'null'
+          title: Is Source Asset
       type: object
       required:
       - source_id
@@ -8079,6 +8084,14 @@ components:
           - type: string
           - type: 'null'
           title: Operator
+        asset_condition_type:
+          anyOf:
+          - type: string
+            enum:
+            - or-gate
+            - and-gate
+          - type: 'null'
+          title: Asset Condition Type
       type: object
       required:
       - id
@@ -9481,6 +9494,9 @@ components:
         dag_id:
           type: string
           title: Dag Id
+        run_id:
+          type: string
+          title: Run Id
       type: object
       required:
       - key
@@ -9489,6 +9505,7 @@ components:
       - map_index
       - task_id
       - dag_id
+      - run_id
       title: XComResponse
       description: Serializer for a xcom item.
     XComResponseNative:
@@ -9513,6 +9530,9 @@ components:
         dag_id:
           type: string
           title: Dag Id
+        run_id:
+          type: string
+          title: Run Id
         value:
           title: Value
       type: object
@@ -9523,6 +9543,7 @@ components:
       - map_index
       - task_id
       - dag_id
+      - run_id
       - value
       title: XComResponseNative
       description: XCom response serializer with native return type.
@@ -9548,6 +9569,9 @@ components:
         dag_id:
           type: string
           title: Dag Id
+        run_id:
+          type: string
+          title: Run Id
         value:
           anyOf:
           - type: string
@@ -9561,6 +9585,7 @@ components:
       - map_index
       - task_id
       - dag_id
+      - run_id
       - value
       title: XComResponseString
       description: XCom response serializer with string return type.

airflow/api_fastapi/core_api/routes/public/connections.py

@@ -121,7 +121,9 @@ def get_connections(
 @connections_router.post(
     "",
     status_code=status.HTTP_201_CREATED,
-    responses=create_openapi_http_exception_doc([status.HTTP_409_CONFLICT]),
+    responses=create_openapi_http_exception_doc(
+        [status.HTTP_409_CONFLICT]
+    ),  # handled by global exception handler
 )
 def post_connection(
     post_body: ConnectionBody,