apache / apache/airflow

AWS deferrable operators drop hook configuration when handing off to triggers

Open
#72,144 13 comments 0 reactions 1 assignee Assigned to @SEPURI-SAI-KRISHNA View on GitHub
area:providers kind:bug provider:amazon
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

> [!IMPORTANT]
> **Status — please read before starting work on this issue.**
>
> Almost everything in the tables below is already covered by an open PR. Please
> comment here before opening a new one, so we don't duplicate each other.
>
> | scope | covered by | state |
> |---|---|---|
> | The `AwsBaseWaiterTrigger` base-class fix, plus the call sites and trigger signatures for bedrock, comprehend, dms, ecs, eks, emr, glue, mwaa, neptune_analytics, rds and ssm | #72171 | open |
> | batch (#72278) | #72449 | open |
> | opensearch_serverless (#72280) | #72472 | open |
> | sagemaker_unified_studio_notebook (#72279) | #72453 | merged |
>
> `SageMakerNotebookOperator` (`aws/operators/sagemaker_unified_studio.py:175`) is
> out of scope: it is a plain `BaseOperator` and carries none of the three
> parameters. The Shape B row for it is a false positive.
>
> That leaves nothing unclaimed. If you would like to help, a review of #72171 is
> worth much more than a new PR right now. If you have already started something,
> please say so here and we will make room for it rather than duplicate it.

### Under which category would you file this issue?

Providers

### Apache Airflow version

3.3.1

### What happened and how to reproduce it?

An `AwsBaseOperator` / `AwsBaseSensor` subclass always carries
`region_name`, `verify` and `botocore_config`, the base class sets all three
in `__init__`. When the task defers, the trigger builds its **own** hook. Any
of the three not explicitly forwarded to the trigger is lost, so the deferred
half of the task talks to AWS with different settings than the synchronous
half:

- **`region_name`** — the triggerer falls back to the boto3 default region, so
the task polls a different region than the one it started work in and waits
on a resource that isn't there.
- **`verify`** — SSL certificate verification silently returns to the default.
A deployment that deliberately set `verify=False`, or pointed it at a
private CA bundle, loses that in the triggerer.
- **`botocore_config`** — custom timeouts, retry policies and connection limits
are dropped, so the triggerer polls with default botocore behaviour.

None of this fails loudly. The task simply behaves differently once it defers.

## Scope

An AST sweep of every `self.defer(trigger=...)` call in the provider finds
**110 defer sites**:

| | sites |
|---|---|
| forward all three | 49 |
| forward only `region_name` | 18 |
| forward none of the three | 43 |
| **remaining to fix** | **61** |

By service:

| service | passes none | passes only `region_name` |
|---|---|---|
| emr | 13 | — |
| bedrock | 12 | — |
| dms | 7 | — |
| comprehend | 4 | — |
| glue | 4 | 2 |
| opensearch_serverless | 1 | — |
| sagemaker_unified_studio | 1 | — |
| sagemaker_unified_studio_notebook | 1 | — |
| eks | — | 9 |
| rds | — | 4 |
| ecs | — | 2 |
| batch | — | 1 |

## Why this should not continue service-by-service

The remaining sites split into two very different shapes:

- **7 sites** can be fixed at the call site, the trigger already names all
three (2 sites), or takes `**kwargs` that reach `AwsBaseWaiterTrigger`
(5 sites).
- **54 sites cannot.** Their trigger subclasses have closed `__init__`
signatures that never accept the parameters, and `hook()` implementations
that build the hook from `aws_conn_id` alone. Fixing those service by
service means editing 54 call sites *and* 54 trigger signatures.

So the bulk of the remaining work is one structural change, not more
per-service PRs.

## Proposed fix

Give `AwsBaseWaiterTrigger` a default `hook()` driven by a `hook_class`
attribute, the same shape as `AwsBaseHookMixin._hook_parameters`, so the
hook is built from the serialized parameters in one place instead of in each
subclass. Add one parametrized invariant test over the deferrable operators
asserting that a deferred trigger's serialized payload carries the operator's
`region_name` / `verify` / `botocore_config`, so a new operator that forgets
them fails in CI rather than in production.

The 7 call-site-only sites can go in the same change or a small follow-up.

## Reproduce

Run any deferrable operator from the table with a non-default `region_name`,
`verify` or `botocore_config`, and inspect the trigger it defers with:

```python
with pytest.raises(TaskDeferred) as deferred:
operator.execute(None)

# the operator's settings are absent from the serialized payload
assert deferred.value.trigger.serialize()[1]
```

## Already fixed

- #71646 — neptune (2 sites)
- #71857 — sagemaker (7 sites)
- #72098 — neptune_analytics, mwaa, ssm (12 sites) — PR open, and also carried by #72171
- #72453 — sagemaker_unified_studio_notebook (1 site)

### What you think should happen instead?

A deferred task should reach AWS with the same region, SSL verification setting and botocore configuration as the synchronous path. Whether a task
defers is an execution detail; it should not change which endpoint the task talks to or how it authenticates to it.

### Operating System

Not Applicable

### Deployment

Other

### Apache Airflow Provider(s)

amazon

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon — audit run against main (9.35.1)

### Official Helm Chart version

Not Applicable

### Kubernetes Version

Not Applicable

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### Anything else?

#### Shape A — trigger already names all three; operator-side fix only (2 sites)

| file:line | operator / sensor | trigger | already passes |
|---|---|---|---|
| `aws/operators/glue.py:361` | `GlueJobOperator` | `GlueJobCompleteTrigger` | region_name |
| `aws/sensors/glue.py:99` | `GlueJobSensor` | `GlueJobCompleteTrigger` | region_name |

#### Shape B — trigger takes `**kwargs`; operator-side fix only (3 sites)

| file:line | operator / sensor | trigger | already passes |
|---|---|---|---|
| `aws/operators/sagemaker_unified_studio.py:175` | `SageMakerNotebookOperator` | `SageMakerNotebookJobTrigger` | — |
| `aws/operators/sagemaker_unified_studio_notebook.py:191` | `SageMakerUnifiedStudioNotebookOperator` | `SageMakerUnifiedStudioNotebookTrigger` | — |
| `aws/sensors/batch.py:99` | `BatchSensor` | `BatchJobTrigger` | region_name |

#### Shape C — trigger signature is closed; needs the base-class fix (55 sites)

| file:line | operator / sensor | trigger | already passes |
|---|---|---|---|
| `aws/operators/bedrock.py:206` | `BedrockCreateAgentRuntimeOperator` | `BedrockAgentRuntimeReadyTrigger` | — |
| `aws/operators/bedrock.py:390` | `BedrockDeleteAgentRuntimeOperator` | `BedrockAgentRuntimeDeletedTrigger` | — |
| `aws/operators/bedrock.py:542` | `BedrockCustomizeModelOperator` | `BedrockCustomizeModelCompletedTrigger` | — |
| `aws/operators/bedrock.py:631` | `BedrockCreateProvisionedModelThroughputOperator` | `BedrockProvisionModelThroughputCompletedTrigger` | — |
| `aws/operators/bedrock.py:804` | `BedrockCreateKnowledgeBaseOperator` | `BedrockKnowledgeBaseActiveTrigger` | — |
| `aws/operators/bedrock.py:989` | `BedrockIngestDataOperator` | `BedrockIngestionJobTrigger` | — |
| `aws/operators/bedrock.py:1308` | `BedrockBatchInferenceOperator` | `BedrockBatchInferenceCompletedTrigger` | — |
| `aws/operators/comprehend.py:189` | `ComprehendStartPiiEntitiesDetectionJobOperator` | `ComprehendPiiEntitiesDetectionJobCompletedTrigger` | — |
| `aws/operators/comprehend.py:338` | `ComprehendCreateDocumentClassifierOperator` | `ComprehendCreateDocumentClassifierCompletedTrigger` | — |
| `aws/operators/dms.py:260` | `DmsModifyTaskOperator` | `DmsTaskModifyCompleteTrigger` | — |
| `aws/operators/dms.py:762` | `DmsDeleteReplicationConfigOperator` | `DmsReplicationTerminalStatusTrigger` | — |
| `aws/operators/dms.py:773` | `DmsDeleteReplicationConfigOperator` | `DmsReplicationDeprovisionedTrigger` | — |
| `aws/operators/dms.py:794` | `DmsDeleteReplicationConfigOperator` | `DmsReplicationConfigDeletedTrigger` | — |
| `aws/operators/dms.py:948` | `DmsStartReplicationOperator` | `DmsReplicationDeprovisionedTrigger` | — |
| `aws/operators/dms.py:991` | `DmsStartReplicationOperator` | `DmsReplicationCompleteTrigger` | — |
| `aws/operators/dms.py:1098` | `DmsStopReplicationOperator` | `DmsReplicationStoppedTrigger` | — |
| `aws/operators/eks.py:358` | `EksCreateClusterOperator` | `EksCreateClusterTrigger` | region_name |
| `aws/operators/eks.py:408` | `EksCreateClusterOperator` | `EksDeleteClusterTrigger` | region_name |
| `aws/operators/eks.py:441` | `EksCreateClusterOperator` | `EksCreateFargateProfileTrigger` | region_name |
| `aws/operators/eks.py:454` | `EksCreateClusterOperator` | `EksCreateNodegroupTrigger` | region_name |
| `aws/operators/eks.py:598` | `EksCreateNodegroupOperator` | `EksCreateNodegroupTrigger` | region_name |
| `aws/operators/eks.py:712` | `EksCreateFargateProfileOperator` | `EksCreateFargateProfileTrigger` | region_name |
| `aws/operators/eks.py:802` | `EksDeleteClusterOperator` | `EksDeleteClusterTrigger` | region_name |
| `aws/operators/eks.py:944` | `EksDeleteNodegroupOperator` | `EksDeleteNodegroupTrigger` | region_name |
| `aws/operators/eks.py:1036` | `EksDeleteFargateProfileOperator` | `EksDeleteFargateProfileTrigger` | region_name |
| `aws/operators/emr.py:234` | `EmrAddStepsOperator` | `EmrAddStepsTrigger` | — |
| `aws/operators/emr.py:865` | `EmrCreateJobFlowOperator` | `EmrCreateJobFlowTrigger` | — |
| `aws/operators/emr.py:1076` | `EmrTerminateJobFlowOperator` | `EmrTerminateJobFlowTrigger` | — |
| `aws/operators/emr.py:1174` | `EmrServerlessCreateApplicationOperator` | `EmrServerlessCreateApplicationTrigger` | — |
| `aws/operators/emr.py:1219` | `EmrServerlessCreateApplicationOperator` | `EmrServerlessStartApplicationTrigger` | — |
| `aws/operators/emr.py:1365` | `EmrServerlessStartJobOperator` | `EmrServerlessStartApplicationTrigger` | — |
| `aws/operators/emr.py:1423` | `EmrServerlessStartJobOperator` | `EmrServerlessStartJobTrigger` | — |
| `aws/operators/emr.py:1682` | `EmrServerlessStopApplicationOperator` | `EmrServerlessCancelJobsTrigger` | — |
| `aws/operators/emr.py:1705` | `EmrServerlessStopApplicationOperator` | `EmrServerlessStopApplicationTrigger` | — |
| `aws/operators/emr.py:1735` | `EmrServerlessStopApplicationOperator` | `EmrServerlessStopApplicationTrigger` | — |
| `aws/operators/emr.py:1828` | `EmrServerlessDeleteApplicationOperator` | `EmrServerlessDeleteApplicationTrigger` | — |
| `aws/operators/glue.py:751` | `GlueDataQualityRuleSetEvaluationRunOperator` | `GlueDataQualityRuleSetEvaluationRunCompleteTrigger` | — |
| `aws/operators/glue.py:899` | `GlueDataQualityRuleRecommendationRunOperator` | `GlueDataQualityRuleRecommendationRunCompleteTrigger` | — |
| `aws/operators/rds.py:646` | `RdsCreateDbInstanceOperator` | `RdsDbAvailableTrigger` | region_name |
| `aws/operators/rds.py:736` | `RdsDeleteDbInstanceOperator` | `RdsDbDeletedTrigger` | region_name |
| `aws/operators/rds.py:820` | `RdsStartDbOperator` | `RdsDbAvailableTrigger` | region_name |
| `aws/operators/rds.py:924` | `RdsStopDbOperator` | `RdsDbStoppedTrigger` | region_name |
| `aws/sensors/bedrock.py:155` | `BedrockCustomizeModelCompletedSensor` | `BedrockCustomizeModelCompletedTrigger` | — |
| `aws/sensors/bedrock.py:225` | `BedrockProvisionModelThroughputCompletedSensor` | `BedrockProvisionModelThroughputCompletedTrigger` | — |
| `aws/sensors/bedrock.py:294` | `BedrockKnowledgeBaseActiveSensor` | `BedrockKnowledgeBaseActiveTrigger` | — |
| `aws/sensors/bedrock.py:381` | `BedrockIngestionJobSensor` | `BedrockIngestionJobTrigger` | — |
| `aws/sensors/comprehend.py:132` | `ComprehendStartPiiEntitiesDetectionJobCompletedSensor` | `ComprehendPiiEntitiesDetectionJobCompletedTrigger` | — |
| `aws/sensors/comprehend.py:217` | `ComprehendCreateDocumentClassifierCompletedSensor` | `ComprehendCreateDocumentClassifierCompletedTrigger` | — |
| `aws/sensors/emr.py:520` | `EmrJobFlowSensor` | `EmrTerminateJobFlowTrigger` | — |
| `aws/sensors/emr.py:648` | `EmrStepSensor` | `EmrStepSensorTrigger` | — |
| `aws/sensors/glue.py:207` | `GlueDataQualityRuleSetEvaluationRunSensor` | `GlueDataQualityRuleSetEvaluationRunCompleteTrigger` | — |
| `aws/sensors/glue.py:327` | `GlueDataQualityRuleRecommendationRunSensor` | `GlueDataQualityRuleRecommendationRunCompleteTrigger` | — |
| `aws/sensors/opensearch_serverless.py:115` | `OpenSearchServerlessCollectionActiveSensor` | `OpenSearchServerlessCollectionActiveTrigger` | — |
| `aws/operators/ecs.py:138` | `EcsCreateClusterOperator` | `ClusterActiveTrigger` | region_name |
| `aws/operators/ecs.py:215` | `EcsDeleteClusterOperator` | `ClusterInactiveTrigger` | region_name |

#### Dynamic trigger selection; needs the base-class fix (1 sites)

| file:line | operator / sensor | trigger | already passes |
|---|---|---|---|
| `aws/sensors/bedrock.py:490` | `BedrockBatchInferenceSensor` | `trigger_class` | — |

The counts come from parsing every `self.defer(trigger=...)` call with `ast`, recording which of the three keywords the trigger construction receives, then reading each trigger class's `__init__` signature to classify the fix shape.

---
Drafted-by: Claude Code (Opus 5); reviewed by @SEPURI-SAI-KRISHNA before posting

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.