Skip to content

Build deferred AWS hooks from the operator's own settings #72171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
potiuk merged 11 commits into from
Sep 21, 2026
Merged

Build deferred AWS hooks from the operator's own settings #72171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
0c3c18b
Use the operator's AWS settings for deferred Neptune, MWAA, SSM tasks
SEPURI-SAI-KRISHNA Aug 26, 2026
3c43a3e
Document the verify parameter for Neptune Analytics operators
SEPURI-SAI-KRISHNA Aug 27, 2026
414e5d8
Build deferred AWS hooks from the operator's own settings
SEPURI-SAI-KRISHNA Aug 27, 2026
26ca9ec
Merge branch 'main' into fix-aws-deferred-hook-config-base
o-nikolas Sep 9, 2026
91e1923
Drop the duplicated Glue trigger arguments after merging main
SEPURI-SAI-KRISHNA Sep 10, 2026
6eed096
Fail fast when an AWS trigger cannot build its hook
SEPURI-SAI-KRISHNA Sep 10, 2026
8acadfe
Treat an unresolvable trigger expression as unreadable
SEPURI-SAI-KRISHNA Sep 18, 2026
dc41980
Resolve the trigger class name where the construction is accepted
SEPURI-SAI-KRISHNA Sep 19, 2026
50ce3e5
Update providers/amazon/tests/unit/amazon/aws/test_deferred_hook_conf…
potiuk Sep 21, 2026
cebb97d
Merge branch 'main' into fix-aws-deferred-hook-config-base
potiuk Sep 21, 2026
32b8d5f
Keep the deferred-hook warning in the pending changelog slot
potiuk Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions providers/amazon/docs/changelog.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,15 @@
Changelog
---------

.. warning::
Comment thread
potiuk marked this conversation as resolved.
Comment thread
potiuk marked this conversation as resolved.
Deferrable AWS operators and sensors now hand ``region_name``, ``verify`` and ``botocore_config``
to the trigger they defer to, so the triggerer builds its hook from the operator's settings
instead of falling back to boto3 defaults. Deployments where the triggerer happened to work
*because* of those defaults will see it change: it now uses the operator's region rather than the
triggerer host's ``AWS_DEFAULT_REGION``, and it applies the operator's SSL verification and
botocore configuration, which previously never reached it. Set these explicitly on the operator
if the deferred half needs to differ from the synchronous half.

9.36.0
......

Expand Down
21 changes: 21 additions & 0 deletions providers/amazon/src/airflow/providers/amazon/aws/operators/bedrock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,9 @@ def execute(self, context: Context) -> str:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -393,6 +396,9 @@ def execute(self, context: Context) -> None:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -545,6 +551,9 @@ def execute(self, context: Context) -> dict:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -634,6 +643,9 @@ def execute(self, context: Context) -> str:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -807,6 +819,9 @@ def _create_kb():
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -994,6 +1009,9 @@ def start_ingestion_job():
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -1311,6 +1329,9 @@ def execute(self, context: Context) -> str:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down
6 changes: 6 additions & 0 deletions providers/amazon/src/airflow/providers/amazon/aws/operators/comprehend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,9 @@ def execute(self, context: Context) -> str:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -341,6 +344,9 @@ def execute(self, context: Context) -> str:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down
21 changes: 21 additions & 0 deletions providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,6 +263,9 @@ def execute(self, context: Context) -> dict:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
kwargs={"result": result},
Expand Down Expand Up @@ -765,6 +768,9 @@ def execute(self, context: Context) -> None:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="retry_execution",
)
Expand All @@ -776,6 +782,9 @@ def execute(self, context: Context) -> None:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="retry_execution",
)
Expand All @@ -797,6 +806,9 @@ def handle_delete_wait(self):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -951,6 +963,9 @@ def execute(self, context: Context):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="retry_execution",
)
Expand Down Expand Up @@ -994,6 +1009,9 @@ def execute(self, context: Context):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down Expand Up @@ -1101,6 +1119,9 @@ def execute(self, context: Context) -> None:
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
botocore_config=self.botocore_config,
),
method_name="execute_complete",
)
Expand Down
9 changes: 8 additions & 1 deletion providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,8 @@ def execute(self, context: Context):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
),
method_name="_complete_exec_with_cluster_desc",
Expand Down Expand Up @@ -218,6 +220,8 @@ def execute(self, context: Context):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
),
method_name="_complete_exec_with_cluster_desc",
Expand Down Expand Up @@ -622,7 +626,10 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
if self._aws_logs_enabled():
# same behavior as non-deferrable mode, return last line of logs of the task.
logs_client = AwsLogsHook(
aws_conn_id=self.aws_conn_id, region_name=self.resolve_awslogs_region()
aws_conn_id=self.aws_conn_id,
region_name=self.resolve_awslogs_region(),
verify=self.verify,
config=self.botocore_config,
).conn
one_log = logs_client.get_log_events(
logGroupName=self.awslogs_group,
Expand Down
18 changes: 18 additions & 0 deletions providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,8 @@ def execute(self, context: Context):
trigger=EksCreateClusterTrigger(
cluster_name=self.cluster_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
Expand Down Expand Up @@ -411,6 +413,8 @@ def deferrable_create_cluster_next(self, context: Context, event: dict[str, Any]
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
force_delete_compute=False,
),
Expand Down Expand Up @@ -445,6 +449,8 @@ def deferrable_create_cluster_next(self, context: Context, event: dict[str, Any]
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
),
method_name="execute_complete",
Expand All @@ -456,6 +462,8 @@ def deferrable_create_cluster_next(self, context: Context, event: dict[str, Any]
nodegroup_name=self.nodegroup_name,
cluster_name=self.cluster_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
Expand Down Expand Up @@ -600,6 +608,8 @@ def execute(self, context: Context):
cluster_name=self.cluster_name,
nodegroup_name=self.nodegroup_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
Expand Down Expand Up @@ -714,6 +724,8 @@ def execute(self, context: Context):
cluster_name=self.cluster_name,
fargate_profile_name=self.fargate_profile_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
region_name=self.region_name,
Expand Down Expand Up @@ -805,6 +817,8 @@ def execute(self, context: Context):
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
force_delete_compute=self.force_delete_compute,
),
Expand Down Expand Up @@ -946,6 +960,8 @@ def execute(self, context: Context):
cluster_name=self.cluster_name,
nodegroup_name=self.nodegroup_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
region_name=self.region_name,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
Expand Down Expand Up @@ -1038,6 +1054,8 @@ def execute(self, context: Context):
cluster_name=self.cluster_name,
fargate_profile_name=self.fargate_profile_name,
aws_conn_id=self.aws_conn_id,
verify=self.verify,
botocore_config=self.botocore_config,
waiter_delay=self.waiter_delay,
waiter_max_attempts=self.waiter_max_attempts,
region_name=self.region_name,
Expand Down
Loading
Loading