AWS deferrable operators drop hook configuration when handing off to triggers

[SEPURI-SAI-KRISHNA]

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:

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

• Use the configured region for deferred Neptune cluster tasks #71646 — neptune (2 sites)
• Use the operator's AWS settings for deferred SageMaker tasks #71857 — sagemaker (7 sites)
• Use the operator's AWS settings for deferred Neptune, MWAA, SSM tasks #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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[jroachgolf84]

@SEPURI-SAI-KRISHNA - does your PR address each of these Operators/Sensors that you mentioned?

[jroachgolf84]

If not, could we use this as part of our Contibutors Workshop to allow for users to make their first contribution?

[rjgoyln]

Hi @SEPURI-SAI-KRISHNA, thanks for driving this!
I ran an AST sweep on main and wanted to share two important findings to help shape the rollout:

1. Critical Rollout Constraint (Crash Risk)
Because the triggerer deserializes via trigger_class(**deserialised_kwargs), we must update the trigger __init__ signatures to accept these parameters before or alongside the operator fixes.
If an operator starts forwarding a real region_name to a trigger that hasn't been updated to accept it, the triggerer will hard-crash with a TypeError (currently, they don't crash because None values are dropped from the payload).

2. Scope Corrections

• Missed (EMR): EmrContainerOperator (line 642) and EmrContainerSensor (line 323) were missed because their triggers are built inside conditional expressions.
• False Positive (SageMaker): SageMakerNotebookOperator is a plain BaseOperator that lacks region_name, verify, or botocore_config. It should be excluded from this scope.
• Signature Count: Only about 30 subclasses actually have fully closed signatures that need fixing, not 54 (e.g., DMS triggers have some partially open signatures).

[SEPURI-SAI-KRISHNA]

Thanks, the EMR catch is real and I've fixed it.

EmrContainerOperator / EmrContainerSensor: you're right, and the miss was worse than one site each. Both build the trigger inside a conditional expression, so there are two constructions per site, four in total, all passing none of the three. My sweep only looked at trigger=<Call> and skipped ast.IfExp entirely, which means my invariant test would have skipped them too. The test now resolves the trigger= expression to every construction it can evaluate to, and a defer site whose trigger is a bare reference is asserted against an explicit allowlist rather than silently ignored. That last part catches one more: EksPodOperator passes a pre-built EksPodTrigger, which is a KubernetesPodTrigger and out of scope.

EmrContainerTrigger also needed its hook's virtual_cluster_id, so it overrides _hook_parameters rather than relying on the default, which is the escape hatch the base class keeps for exactly this.

SageMakerNotebookOperator: agreed, excluded. I'd reached the same conclusion from the trigger side, SageMakerNotebookJobTrigger is a plain BaseTrigger whose hook is addressed by execution name, and your reason from the operator side is the same boundary.

Ordering / crash risk: agreed, and it's why this is one PR rather than a series. The trigger signatures and the call sites move together, so there is no intermediate commit where an operator forwards a parameter its trigger cannot accept.

Signature count: we're measuring different things, and your number is the right one for classes. My 54 was defer sites that could not be fixed at the call site. Counting classes instead: of the 45 this PR touches, 31 accept none of the three, 12 already accept region_name, and 2 accept some other subset. So ~30 fully closed is correct; I should have said "sites" more clearly in the issue.

Net effect after the fix: 113 of 114 trigger constructions forward the hook configuration, the remaining one being the SageMaker case above.

---

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

[SEPURI-SAI-KRISHNA]

It does, yes, I have it working locally across all of them: 113 of 114 defer sites, the one exception being SageMakerNotebookOperator, which is a different shape and out of scope.

But I'd rather not take the whole issue off the table for the workshop, and it splits cleanly if that's more useful to you.

The reason the remaining sites can't just be fixed one call at a time is that most of the trigger subclasses never accept region_name / verify / botocore_config at all, and each builds its own hook from aws_conn_id alone. So I'd propose landing only the mechanism:

• AwsBaseWaiterTrigger builds the hook itself from the parameters it already serializes, driven by an aws_hook_class attribute, the same arrangement AwsBaseHookMixin gives the operators.
• An invariant test that walks every defer site in the provider and fails if one doesn't forward its hook configuration, carrying an allowlist of the services not yet migrated.

That's roughly 200 lines and no behaviour change on its own. Each service then becomes a self-contained first contribution: set aws_hook_class on the triggers, widen their __init__, pass the three parameters at the call sites, delete one entry from the allowlist and watch that service's tests go green.

Sized by defer sites and trigger classes:

service	sites	trigger classes
emr	17	11
bedrock	12	8
eks	9	6
dms	7	6
glue	6	3
comprehend	4	2
rds	4	3
ecs	2	2
batch	1	1
opensearch_serverless	1	1
sagemaker_unified_studio_notebook	1	1

The small ones at the bottom are a gentle first PR; emr and bedrock are better suited to someone who has already done one.

Happy either way, say the word and I'll open it as the mechanism only, or as the complete fix if you'd rather just have it done.

---

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

[jroachgolf84]

@SEPURI-SAI-KRISHNA - thanks for the write up! Do you mind if I steal batch, opensearch_serverless, and sagemarker_unified_studio_notebook for the Contributor's Workshop? I will create sub-issues for them if it's alright by you.

Really appreciate this! The Airflow Community thanks you!