SEPURI-SAI-KRISHNA opened a new issue, #72144:
URL: https://github.com/apache/airflow/issues/72144
### 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)
### 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 (5 sites)
| file:line | operator / sensor | trigger | already passes |
|---|---|---|---|
| `aws/operators/ecs.py:138` | `EcsCreateClusterOperator` |
`ClusterActiveTrigger` | region_name |
| `aws/operators/ecs.py:215` | `EcsDeleteClusterOperator` |
`ClusterInactiveTrigger` | region_name |
| `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 (53
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` | — |
#### 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
