Don't allow passing include_deferred at all if global conf is set to true/false
diff --git a/airflow-core/docs/administration-and-deployment/pools.rst b/airflow-core/docs/administration-and-deployment/pools.rst index 2c15963..063d45e 100644 --- a/airflow-core/docs/administration-and-deployment/pools.rst +++ b/airflow-core/docs/administration-and-deployment/pools.rst
@@ -49,8 +49,10 @@ Whether deferred tasks occupy pool slots is normally decided per pool via its ``include_deferred`` flag. A Deployment Manager can instead fix this behavior for the whole cluster with :ref:`config:core__pool_include_deferred`. When that option is set to ``True`` or ``False``, the configured -value applies to every pool (including pre-existing pools, regardless of their stored flag), and attempts -to explicitly set a conflicting ``include_deferred`` value when creating or updating a pool are rejected. +value applies to every pool (including pre-existing pools, regardless of their stored flag), and any attempt +to set ``include_deferred`` when creating or updating a pool is rejected — whether via the API or the CLI, +and whether or not the requested value matches the configured one. The field is read-only in the UI, which +omits it from create and update requests entirely. Using multiple pool slots -------------------------
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py index df717b2..8528e62 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py
@@ -41,12 +41,21 @@ return value if override is None else override -def _reject_conflicting_include_deferred(value: bool, override: bool) -> None: - if value != override: +def _reject_user_provided_include_deferred(provided_fields: set[str]) -> bool | None: + """ + Get the cluster-wide ``include_deferred`` value, rejecting any value provided by the user. + + :param provided_fields: names of the fields explicitly present in the request body. + :return: the cluster-wide value, or None when pools choose the flag themselves. + :raises ValueError: if the flag is fixed cluster-wide but present in the request body. + """ + override = Pool.get_include_deferred_override() + if override is not None and "include_deferred" in provided_fields: raise ValueError( - f"include_deferred is fixed to {override} for all pools by the [core] pool_include_deferred " - "configuration and cannot be set per pool. Please contact your administrator." + f"include_deferred cannot be set because it is fixed to {override} for all pools by the " + "[core] pool_include_deferred configuration. Please contact your administrator." ) + return override PoolSlots = Annotated[ @@ -111,9 +120,7 @@ @model_validator(mode="after") def enforce_include_deferred_override(self) -> PoolPatchBody: - override = Pool.get_include_deferred_override() - if override is not None and self.include_deferred is not None: - _reject_conflicting_include_deferred(self.include_deferred, override) + _reject_user_provided_include_deferred(self.model_fields_set) return self @@ -135,10 +142,7 @@ @model_validator(mode="after") def enforce_include_deferred_override(self) -> PoolBody: - override = Pool.get_include_deferred_override() - if override is None: - return self - if "include_deferred" in self.model_fields_set: - _reject_conflicting_include_deferred(self.include_deferred, override) - self.include_deferred = override + override = _reject_user_provided_include_deferred(self.model_fields_set) + if override is not None: + self.include_deferred = override return self
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py b/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py index 476c45b..75f7f6f 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py
@@ -93,6 +93,11 @@ if "name" in body_dict and "pool" not in body_dict: body_dict["pool"] = body_dict.pop("name") + # When fixed cluster-wide the client must not send include_deferred, so supply the + # server-side value to satisfy BasePool, which requires it + if (include_deferred_override := Pool.get_include_deferred_override()) is not None: + body_dict["include_deferred"] = include_deferred_override + BasePool.model_validate(body_dict) except ValidationError as e:
diff --git a/airflow-core/src/airflow/cli/commands/pool_command.py b/airflow-core/src/airflow/cli/commands/pool_command.py index 000349a..1f9953a 100644 --- a/airflow-core/src/airflow/cli/commands/pool_command.py +++ b/airflow-core/src/airflow/cli/commands/pool_command.py
@@ -33,15 +33,15 @@ from airflow.utils.providers_configuration_loader import providers_configuration_loaded -def check_include_deferred_choice_allowed(include_deferred: bool) -> str | None: - """Return an error message when an explicit ``include_deferred`` choice conflicts with the cluster config.""" +def get_include_deferred_rejection() -> str | None: + """Return an error message when ``include_deferred`` is fixed cluster-wide and so cannot be set by the user.""" override = Pool.get_include_deferred_override() - if override is not None and include_deferred != override: - return ( - f"include_deferred is fixed to {override} for all pools by the [core] pool_include_deferred " - "configuration and cannot be set per pool." - ) - return None + if override is None: + return None + return ( + f"include_deferred cannot be set because it is fixed to {override} for all pools by the " + "[core] pool_include_deferred configuration." + ) def _show_pools(pools, output): @@ -87,8 +87,8 @@ @providers_configuration_loaded def pool_set(args): """Create new pool with a given name and slots.""" - # --include-deferred is a store-true flag, so only a passed flag is an explicit choice - if args.include_deferred and (error := check_include_deferred_choice_allowed(True)): + # --include-deferred is a store-true flag, so only a passed flag is a value provided by the user + if args.include_deferred and (error := get_include_deferred_rejection()): raise SystemExit(error) api_client = get_current_api_client() api_client.create_pool( @@ -151,9 +151,7 @@ failed = [] for k, v in pools_json.items(): if isinstance(v, dict) and "slots" in v and "description" in v: - if "include_deferred" in v and ( - error := check_include_deferred_choice_allowed(bool(v["include_deferred"])) - ): + if "include_deferred" in v and (error := get_include_deferred_rejection()): print(f"Pool {k}: {error}") failed.append(k) continue @@ -174,14 +172,17 @@ def pool_export_helper(filepath): """Help export all the pools to the json file.""" api_client = get_current_api_client() + # Omit the field when it is fixed cluster-wide, otherwise the exported file cannot be imported back + export_include_deferred = Pool.get_include_deferred_override() is None pool_dict = {} pools = api_client.get_pools() for pool in pools: entry = { "slots": pool[1], "description": pool[2], - "include_deferred": pool[3], } + if export_include_deferred: + entry["include_deferred"] = pool[3] if pool[4] is not None: entry["team_name"] = pool[4] pool_dict[pool[0]] = entry
diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index cb9199b..84e3ae7 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -480,8 +480,8 @@ each pool keeps its own ``include_deferred`` value, configurable per pool via the UI, API or CLI. When set to ``True`` or ``False``, the configured value is used for **every** pool (including pre-existing pools, whatever their stored value) when calculating occupied slots, and users can - no longer choose the flag per pool: attempts to explicitly set a conflicting ``include_deferred`` - value when creating or updating a pool are rejected. + no longer choose the flag per pool: any attempt to set ``include_deferred`` when creating or + updating a pool is rejected, whether or not the requested value matches the configured one. version_added: 3.4.0 type: string example: "True"
diff --git a/airflow-core/src/airflow/ui/src/pages/Pools/PoolForm.tsx b/airflow-core/src/airflow/ui/src/pages/Pools/PoolForm.tsx index 484bd1d7..82f35ba 100644 --- a/airflow-core/src/airflow/ui/src/pages/Pools/PoolForm.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Pools/PoolForm.tsx
@@ -28,7 +28,8 @@ export type PoolBody = { description: string | undefined; - include_deferred: boolean; + // Left undefined when the value is fixed cluster-wide, so it is not sent to the API + include_deferred: boolean | undefined; name: string; slots: number; team_name: string; @@ -62,9 +63,7 @@ typeof includeDeferredConfig === "boolean" ? includeDeferredConfig : undefined; const onSubmit = (data: PoolBody) => { - manageMutate( - includeDeferredOverride === undefined ? data : { ...data, include_deferred: includeDeferredOverride }, - ); + manageMutate(includeDeferredOverride === undefined ? data : { ...data, include_deferred: undefined }); }; const handleReset = () => {
diff --git a/airflow-core/src/airflow/ui/src/queries/useEditPool.ts b/airflow-core/src/airflow/ui/src/queries/useEditPool.ts index 44075eb..2aa82de 100644 --- a/airflow-core/src/airflow/ui/src/queries/useEditPool.ts +++ b/airflow-core/src/airflow/ui/src/queries/useEditPool.ts
@@ -74,7 +74,10 @@ parsedDescription = editPoolRequestBody.description; updateMask.push("description"); } - if (editPoolRequestBody.include_deferred !== initialPool.include_deferred) { + if ( + editPoolRequestBody.include_deferred !== undefined && + editPoolRequestBody.include_deferred !== initialPool.include_deferred + ) { updateMask.push("include_deferred"); } if (editPoolRequestBody.team_name !== initialPool.team_name) {
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py index feecee6..a2a903f 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py
@@ -468,22 +468,30 @@ assert response.json() == expected_response check_last_log(session, dag_id=None, event="patch_pool", logical_date=None) - def test_patch_pool_with_cluster_fixed_include_deferred(self, test_client): + @pytest.mark.parametrize("include_deferred", [True, False, None]) + def test_patch_pool_rejects_include_deferred_when_fixed_by_config(self, test_client, include_deferred): self.create_pools() with conf_vars({("core", "pool_include_deferred"): "False"}): - # conflicting explicit value is rejected response = test_client.patch( f"/pools/{POOL2_NAME}", - json={"name": POOL2_NAME, "slots": 5, "include_deferred": True}, + json={"name": POOL2_NAME, "slots": 5, "include_deferred": include_deferred}, ) - assert response.status_code == 422 - assert "include_deferred is fixed to False" in response.json()["detail"][0]["msg"] - # matching value is accepted + assert response.status_code == 422 + assert ( + "include_deferred cannot be set because it is fixed to False" + in response.json()["detail"][0]["msg"] + ) + + def test_patch_pool_without_include_deferred_when_fixed_by_config(self, test_client, session): + self.create_pools() + with conf_vars({("core", "pool_include_deferred"): "True"}): response = test_client.patch( f"/pools/{POOL2_NAME}", - json={"name": POOL2_NAME, "slots": 5, "include_deferred": False}, + json={"name": POOL2_NAME, "slots": 5}, ) - assert response.status_code == 200 + assert response.status_code == 200 + assert response.json()["include_deferred"] is True + assert response.json()["slots"] == 5 @conf_vars({("core", "multi_team"): "False"}) def test_patch_pool_rejects_team_name_when_multi_team_disabled(self, test_client): @@ -612,38 +620,31 @@ ) assert response.status_code == 422 - @pytest.mark.parametrize( - ("conf_value", "body_include_deferred", "expected_status_code", "expected_include_deferred"), - [ - ("True", None, 201, True), - ("True", True, 201, True), - ("False", True, 422, None), - ("True", False, 422, None), - ], - ) - def test_post_pool_with_cluster_fixed_include_deferred( - self, - test_client, - session, - conf_value, - body_include_deferred, - expected_status_code, - expected_include_deferred, + @pytest.mark.parametrize("conf_value", ["True", "False"]) + @pytest.mark.parametrize("body_include_deferred", [True, False]) + def test_post_pool_rejects_include_deferred_when_fixed_by_config( + self, test_client, conf_value, body_include_deferred ): - body = {"name": "locked_pool", "slots": 1} - if body_include_deferred is not None: - body["include_deferred"] = body_include_deferred with conf_vars({("core", "pool_include_deferred"): conf_value}): - response = test_client.post("/pools", json=body) - assert response.status_code == expected_status_code - if expected_status_code == 201: - assert response.json()["include_deferred"] is expected_include_deferred - assert ( - session.scalar(select(Pool.include_deferred).where(Pool.pool == "locked_pool")) - is expected_include_deferred + response = test_client.post( + "/pools", + json={"name": "locked_pool", "slots": 1, "include_deferred": body_include_deferred}, ) - else: - assert "include_deferred is fixed to" in response.json()["detail"][0]["msg"] + assert response.status_code == 422 + assert ( + f"include_deferred cannot be set because it is fixed to {conf_value}" + in response.json()["detail"][0]["msg"] + ) + + @pytest.mark.parametrize(("conf_value", "expected"), [("True", True), ("False", False)]) + def test_post_pool_without_include_deferred_stores_config_value( + self, test_client, session, conf_value, expected + ): + with conf_vars({("core", "pool_include_deferred"): conf_value}): + response = test_client.post("/pools", json={"name": "locked_pool", "slots": 1}) + assert response.status_code == 201 + assert response.json()["include_deferred"] is expected + assert session.scalar(select(Pool.include_deferred).where(Pool.pool == "locked_pool")) is expected @conf_vars({("core", "multi_team"): "False"}) def test_post_pool_rejects_team_name_when_multi_team_disabled(self, test_client): @@ -1227,6 +1228,31 @@ assert updated_pool.description is None # unchanged assert updated_pool.include_deferred is True # unchanged + def test_bulk_create_rejects_include_deferred_when_fixed_by_config(self, test_client): + request_body = { + "actions": [ + { + "action": "create", + "entities": [{"name": "locked_pool", "slots": 1, "include_deferred": True}], + } + ] + } + with conf_vars({("core", "pool_include_deferred"): "True"}): + response = test_client.patch("/pools", json=request_body) + assert response.status_code == 422 + assert ( + "include_deferred cannot be set because it is fixed to True" + in response.json()["detail"][0]["msg"] + ) + + def test_bulk_create_without_include_deferred_stores_config_value(self, test_client, session): + request_body = {"actions": [{"action": "create", "entities": [{"name": "locked_pool", "slots": 1}]}]} + with conf_vars({("core", "pool_include_deferred"): "True"}): + response = test_client.patch("/pools", json=request_body) + assert response.status_code == 200 + assert response.json()["create"]["success"] == ["locked_pool"] + assert session.scalar(select(Pool.include_deferred).where(Pool.pool == "locked_pool")) is True + @pytest.mark.parametrize( ("pool_count"), [5, 10, 20],
diff --git a/airflow-core/tests/unit/cli/commands/test_pool_command.py b/airflow-core/tests/unit/cli/commands/test_pool_command.py index 7abcf52..b6e7f0f 100644 --- a/airflow-core/tests/unit/cli/commands/test_pool_command.py +++ b/airflow-core/tests/unit/cli/commands/test_pool_command.py
@@ -81,27 +81,31 @@ pool_command.pool_set(self.parser.parse_args(["pools", "set", "foo", "1", "test"])) assert self.session.scalar(select(Pool).where(Pool.pool == "foo")).include_deferred is False - def test_pool_set_include_deferred_rejected_when_fixed_by_config(self): - with conf_vars({("core", "pool_include_deferred"): "False"}): - with pytest.raises(SystemExit, match="include_deferred is fixed to False for all pools"): + @pytest.mark.parametrize("conf_value", ["True", "False"]) + def test_pool_set_include_deferred_rejected_when_fixed_by_config(self, conf_value): + with conf_vars({("core", "pool_include_deferred"): conf_value}): + with pytest.raises( + SystemExit, match=f"include_deferred cannot be set because it is fixed to {conf_value}" + ): pool_command.pool_set( self.parser.parse_args(["pools", "set", "locked_pool", "1", "test", "--include-deferred"]) ) assert self.session.scalar(select(Pool).where(Pool.pool == "locked_pool")) is None - def test_pool_set_include_deferred_allowed_when_matching_config(self): + def test_pool_set_without_include_deferred_stores_config_value(self): try: with conf_vars({("core", "pool_include_deferred"): "True"}): - pool_command.pool_set( - self.parser.parse_args(["pools", "set", "locked_pool", "1", "test", "--include-deferred"]) - ) + pool_command.pool_set(self.parser.parse_args(["pools", "set", "locked_pool", "1", "test"])) assert self.session.scalar(select(Pool).where(Pool.pool == "locked_pool")).include_deferred finally: self._cleanup() - def test_pool_import_include_deferred_rejected_when_fixed_by_config(self, tmp_path): + @pytest.mark.parametrize("file_value", [True, False]) + def test_pool_import_include_deferred_rejected_when_fixed_by_config(self, tmp_path, file_value): pool_import_file_path = tmp_path / "pools_import.json" - pool_config_input = {"locked_pool": {"slots": 1, "description": "test", "include_deferred": True}} + pool_config_input = { + "locked_pool": {"slots": 1, "description": "test", "include_deferred": file_value} + } with open(pool_import_file_path, mode="w") as file: json.dump(pool_config_input, file) @@ -112,6 +116,25 @@ ) assert self.session.scalar(select(Pool).where(Pool.pool == "locked_pool")) is None + def test_pool_export_import_omits_include_deferred_when_fixed_by_config(self, tmp_path): + pool_export_file_path = tmp_path / "pools_export.json" + try: + pool_command.pool_set(self.parser.parse_args(["pools", "set", "locked_pool", "1", "test"])) + with conf_vars({("core", "pool_include_deferred"): "True"}): + pool_command.pool_export( + self.parser.parse_args(["pools", "export", str(pool_export_file_path)]) + ) + with open(pool_export_file_path) as file: + exported = json.load(file) + assert "include_deferred" not in exported["locked_pool"] + + # the exported file can be imported back without being rejected + pool_command.pool_import( + self.parser.parse_args(["pools", "import", str(pool_export_file_path)]) + ) + finally: + self._cleanup() + def test_pool_get(self): pool_command.pool_set(self.parser.parse_args(["pools", "set", "foo", "1", "test"])) pool_command.pool_get(self.parser.parse_args(["pools", "get", "foo"]))