This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 ed6344d9dda Deprecate deferrable overriding wait_for_completion in
Comprehend (#75226)
ed6344d9dda is described below
commit ed6344d9dda462735a4566278384797750934ed4
Author: Ramit Kataria <[email protected]>
AuthorDate: Wed Oct 21 10:40:26 2026 -0700
Deprecate deferrable overriding wait_for_completion in Comprehend (#75226)
Most Amazon provider operators that accept both ``deferrable`` and
``wait_for_completion`` defer whenever ``deferrable`` is True, so an
explicit ``wait_for_completion=False`` is ignored. Because ``deferrable``
defaults to the ``[operators] default_deferrable`` option, a
deployment-wide setting overrides a task-level choice not to wait. A
start operator paired with its companion sensor then waits twice, and the
operator's shorter waiter can fail the task before the sensor's budget
runs out. The Comprehend system test hit this.
The next major release will defer only when ``wait_for_completion`` is
True, matching the SageMaker Processing/Training/Transform, DMS, Redshift
Resume/Pause and EMR job flow operators. Warning about the affected
combination now lets Dag authors pin either behavior before it changes.
Comprehend goes first so the shared decision point and warning wording
can be settled on one service before the rest of the provider follows.
---
.../providers/amazon/aws/operators/comprehend.py | 13 ++++--
.../airflow/providers/amazon/aws/utils/deferral.py | 41 +++++++++++++++++
.../unit/amazon/aws/operators/test_comprehend.py | 17 ++++---
.../tests/unit/amazon/aws/utils/test_deferral.py | 53 ++++++++++++++++++++++
4 files changed, 114 insertions(+), 10 deletions(-)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/comprehend.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/comprehend.py
index 6cdf3e175b5..65395c8b359 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/comprehend.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/comprehend.py
@@ -31,6 +31,7 @@ from airflow.providers.amazon.aws.triggers.comprehend import (
ComprehendPiiEntitiesDetectionJobCompletedTrigger,
)
from airflow.providers.amazon.aws.utils import validate_execute_complete_event
+from airflow.providers.amazon.aws.utils.deferral import should_defer
from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
from airflow.providers.common.compat.sdk import AirflowException, conf,
timezone
@@ -108,7 +109,9 @@ class
ComprehendStartPiiEntitiesDetectionJobOperator(ComprehendBaseOperator):
:param waiter_delay: Time in seconds to wait between status checks.
(default: 60)
:param waiter_max_attempts: Maximum number of attempts to check for job
completion. (default: 20)
:param deferrable: If True, the operator will wait asynchronously for the
job to stop.
- This implies waiting for completion. This mode requires aiobotocore
module to be installed.
+ Deferring when ``wait_for_completion`` is False is deprecated; from
the next major release
+ this has no effect when ``wait_for_completion`` is False.
+ This mode requires aiobotocore module to be installed.
(default: False)
:param aws_conn_id: The Airflow connection used for AWS credentials.
If this is ``None`` or empty then the default boto3 behaviour is used.
If
@@ -184,7 +187,7 @@ class
ComprehendStartPiiEntitiesDetectionJobOperator(ComprehendBaseOperator):
)
self.log.info("You can view the PII entities detection job at %s",
job_url)
message_description = f"start pii entities detection job {job_id} to
complete."
- if self.deferrable:
+ if should_defer(deferrable=self.deferrable,
wait_for_completion=self.wait_for_completion):
self.log.info("Deferring %s", message_description)
self.defer(
trigger=ComprehendPiiEntitiesDetectionJobCompletedTrigger(
@@ -243,7 +246,9 @@ class
ComprehendCreateDocumentClassifierOperator(AwsBaseOperator[ComprehendHook]
:param waiter_delay: Time in seconds to wait between status checks.
(default: 60)
:param waiter_max_attempts: Maximum number of attempts to check for job
completion. (default: 60)
:param deferrable: If True, the operator will wait asynchronously for the
job to stop.
- This implies waiting for completion. This mode requires aiobotocore
module to be installed.
+ Deferring when ``wait_for_completion`` is False is deprecated; from
the next major release
+ this has no effect when ``wait_for_completion`` is False.
+ This mode requires aiobotocore module to be installed.
(default: False)
:param aws_conn_id: The Airflow connection used for AWS credentials.
If this is ``None`` or empty then the default boto3 behaviour is used.
If
@@ -336,7 +341,7 @@ class
ComprehendCreateDocumentClassifierOperator(AwsBaseOperator[ComprehendHook]
self.log.info("You can monitor the classifier at %s", job_url)
message_description = f"document classifier {document_classifier_arn}
to complete."
- if self.deferrable:
+ if should_defer(deferrable=self.deferrable,
wait_for_completion=self.wait_for_completion):
self.log.info("Deferring %s", message_description)
self.defer(
trigger=ComprehendCreateDocumentClassifierCompletedTrigger(
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/utils/deferral.py
b/providers/amazon/src/airflow/providers/amazon/aws/utils/deferral.py
new file mode 100644
index 00000000000..11d6bc69c0f
--- /dev/null
+++ b/providers/amazon/src/airflow/providers/amazon/aws/utils/deferral.py
@@ -0,0 +1,41 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import warnings
+
+from airflow.exceptions import AirflowProviderDeprecationWarning
+
+DEFERRABLE_OVERRIDES_WAIT_WARNING = (
+ "deferrable=True currently takes precedence over
wait_for_completion=False, so this task still "
+ "waits for completion in the triggerer. In the next major release of the
Amazon provider, the task "
+ "will not wait when wait_for_completion is False. deferrable may come from
the [operators] "
+ "default_deferrable option. To keep waiting, set wait_for_completion=True.
To stop waiting now, "
+ "set deferrable=False."
+)
+
+
+def should_defer(*, deferrable: bool, wait_for_completion: bool) -> bool:
+ """
+ Decide whether an operator that accepts both ``deferrable`` and
``wait_for_completion`` defers.
+
+ Deferring still takes precedence over ``wait_for_completion=False``, with
a deprecation warning.
+ In the next major release this returns ``deferrable and
wait_for_completion``.
+ """
+ if deferrable and not wait_for_completion:
+ warnings.warn(DEFERRABLE_OVERRIDES_WAIT_WARNING,
AirflowProviderDeprecationWarning, stacklevel=2)
+ return deferrable
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_comprehend.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_comprehend.py
index c9ed74b9e5d..7c66d5222cd 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_comprehend.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_comprehend.py
@@ -31,6 +31,7 @@ from airflow.providers.amazon.aws.operators.comprehend import
(
)
from airflow.providers.amazon.version_compat import NOTSET
+from unit.amazon.aws.utils.test_deferral import expect_deferral_warning
from unit.amazon.aws.utils.test_template_fields import validate_template_fields
if TYPE_CHECKING:
@@ -173,7 +174,8 @@ class TestComprehendStartPiiEntitiesDetectionJobOperator:
[
pytest.param(False, False, id="no_wait"),
pytest.param(True, False, id="wait"),
- pytest.param(False, True, id="defer"),
+ pytest.param(True, True, id="defer"),
+ pytest.param(False, True, id="defer_overrides_no_wait"),
],
)
@mock.patch.object(ComprehendHook, "get_waiter")
@@ -183,10 +185,11 @@ class TestComprehendStartPiiEntitiesDetectionJobOperator:
self.operator.wait_for_completion = wait_for_completion
self.operator.deferrable = deferrable
- response = self.operator.execute({})
+ with expect_deferral_warning(deferrable=deferrable,
wait_for_completion=wait_for_completion):
+ response = self.operator.execute({})
assert response == self.JOB_ID
- assert comprehend_hook.get_waiter.call_count == wait_for_completion
+ assert comprehend_hook.get_waiter.call_count == (wait_for_completion
and not deferrable)
assert self.operator.defer.call_count == deferrable
def test_template_fields(self):
@@ -275,7 +278,8 @@ class TestComprehendCreateDocumentClassifierOperator:
[
pytest.param(False, False, id="no_wait"),
pytest.param(True, False, id="wait"),
- pytest.param(False, True, id="defer"),
+ pytest.param(True, True, id="defer"),
+ pytest.param(False, True, id="defer_overrides_no_wait"),
],
)
@mock.patch.object(ComprehendHook, "get_waiter")
@@ -285,10 +289,11 @@ class TestComprehendCreateDocumentClassifierOperator:
self.operator.wait_for_completion = wait_for_completion
self.operator.deferrable = deferrable
- response = self.operator.execute({})
+ with expect_deferral_warning(deferrable=deferrable,
wait_for_completion=wait_for_completion):
+ response = self.operator.execute({})
assert response == self.CLASSIFIER_ARN
- assert comprehend_hook.get_waiter.call_count == wait_for_completion
+ assert comprehend_hook.get_waiter.call_count == (wait_for_completion
and not deferrable)
assert self.operator.defer.call_count == deferrable
def test_template_fields(self):
diff --git a/providers/amazon/tests/unit/amazon/aws/utils/test_deferral.py
b/providers/amazon/tests/unit/amazon/aws/utils/test_deferral.py
new file mode 100644
index 00000000000..0679ab80537
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/utils/test_deferral.py
@@ -0,0 +1,53 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import warnings
+from contextlib import AbstractContextManager, nullcontext
+
+import pytest
+
+from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.amazon.aws.utils.deferral import should_defer
+
+
+def expect_deferral_warning(*, deferrable: bool, wait_for_completion: bool) ->
AbstractContextManager:
+ """Expect the ``should_defer`` deprecation warning only for the
combination that triggers it."""
+ if deferrable and not wait_for_completion:
+ return pytest.warns(
+ AirflowProviderDeprecationWarning, match="takes precedence over
wait_for_completion"
+ )
+ return nullcontext()
+
+
[email protected](
+ ("deferrable", "wait_for_completion", "expected"),
+ [
+ pytest.param(False, False, False, id="no_wait"),
+ pytest.param(False, True, False, id="wait"),
+ pytest.param(True, True, True, id="defer"),
+ ],
+)
+def test_should_defer_without_warning(deferrable, wait_for_completion,
expected):
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ assert should_defer(deferrable=deferrable,
wait_for_completion=wait_for_completion) is expected
+
+
+def test_should_defer_warns_when_deferrable_overrides_wait():
+ with expect_deferral_warning(deferrable=True, wait_for_completion=False):
+ assert should_defer(deferrable=True, wait_for_completion=False) is True
