Forward hook config to `BatchJobTrigger` from the `BatchSensor`
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
## Description
> This is an issue that is reserved for the Airflow Summit "Contributors" Workshop. This is denoted with the label `contributors-workshop`. Out of respect for the organizers and participants of this workshop, **please do not implement a PR that addresses this issue.**
>
> If this issue is still open following Airflow Summit, the label will be removed and the issue can be picked up.
Sub-issue of #72144, covering one of the three defer sites set aside for the Contributor's Workshop in [this comment](https://github.com/apache/airflow/issues/72144#issuecomment-5443098667).
| | |
| --- | --- |
| Call site | `providers/amazon/src/airflow/providers/amazon/aws/sensors/batch.py`, in `BatchSensor.execute` |
| Trigger | `BatchJobTrigger` in `providers/amazon/src/airflow/providers/amazon/aws/triggers/batch.py` |
| Shape | Call site only, the trigger already accepts the parameters |
`BatchSensor` is an `AwsBaseSensor`, so it always carries `region_name`, `verify` and `botocore_config`. When it defers it passes only `region_name` to `BatchJobTrigger`, so the triggerer rebuilds its hook without the SSL verification setting and without the botocore configuration the synchronous half of the task was using.
### What needs to change?
This one is small because nothing on the trigger side is in the way. `BatchJobTrigger.__init__` already takes `**kwargs` and forwards them to `AwsBaseWaiterTrigger`, and its `hook()` already reads all three values off `self`:
```python
def hook(self) -> AwsGenericHook:
return BatchClientHook(
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
verify=self.verify,
config=self.botocore_config,
)
```
They are simply never populated, because the sensor stops after `region_name`. The fix is to add the two missing arguments at the defer site so the trigger receives what it is already prepared to use.
There is a working reference for this in the same provider. `BatchOperator` defers to the same trigger from `providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py` and already passes all three, so the sensor should end up matching that call.
### How to verify it
The parent issue's reproduction applies: construct the sensor with a non-default `region_name`, `verify` and `botocore_config`, trigger the defer, and assert those values survive into the serialized payload.
```python
with pytest.raises(TaskDeferred) as deferred:
sensor.execute(None)
assert deferred.value.trigger.serialize()[1]
```
`AwsBaseWaiterTrigger.serialize` prunes empty values, so use values that are actually distinguishable from the defaults, and assert on the serialized dict rather than on attributes of the trigger object.
## Definition of Done
1. Pass `verify` and `botocore_config` through to `BatchJobTrigger` at the defer site in `sensors/batch.py`.
2. Add or extend a unit test asserting all three survive into the serialized trigger payload.
3. Once #72171 has landed, remove the `("sensors/batch.py", "BatchJobTrigger")` entry from the `PENDING_MIGRATION` allowlist that PR introduces. That allowlist does not exist on `main` yet, so this step only applies after it merges. The invariant test asserts each entry is still needed, so a stale line fails the suite.
4. These should pass:
```bash
breeze testing providers-tests providers/amazon/tests/unit/amazon/aws/sensors/test_batch.py
breeze testing providers-tests providers/amazon/tests/unit/amazon/aws/triggers/test_batch.py
```
Note on sequencing. The code change here is independent of #72171 and can be made and reviewed straight away. Only the allowlist deletion in step 3 has to wait for that PR.
---
Drafted-by: Claude Code (Opus 5); reviewed and edited by @jroachgolf84 before posting
Contributor guide
Assessment
This issue has not been assessed yet.