haeganm opened a new pull request, #72770:
URL: https://github.com/apache/airflow/pull/72770
`AppflowHook.run_flow` ends by calling `_log_execution_description`, which
looked up the execution it just ran in a single unpaginated
`DescribeFlowExecutionRecords` call:
```python
response_desc = self.conn.describe_flow_execution_records(flowName=flow_name)
last_execs = {fe["executionId"]: fe for fe in
response_desc["flowExecutions"]}
exec_details = last_execs[execution_id]
```
`maxResults` defaults to 20 and the response carries a `nextToken`, and the
API does not document any ordering, so the execution is not guaranteed to be on
that first page. When it is not, that dict lookup raises `KeyError` and the
task fails even though AppFlow reported the flow as complete.
`wait_for_completion` defaults to `True`, so every AppFlow operator goes
through this.
`AppflowRecordsShortCircuitOperator._has_new_records_func` already treats
the same API as paginated, so this brings the hook in line with it.
### What changed
* `_log_execution_description` walks pages until it finds the execution,
asking for 100 records per page, which is the API maximum. In practice that is
one call.
* If no record is found on any page it logs a warning instead of raising.
This is a behavior change and it is deliberate: the helper only exists to log
the execution details, so it should not be able to fail a run that AppFlow
already reported as successful.
* The waiter in `run_flow` reads the same API, so it now asks for 100
records per page as well. Botocore waiters cannot paginate, so this widens the
window rather than closing it. Called out below.
* `MAX_EXECUTION_RECORDS_PER_PAGE = 100` so both call sites use one value.
### Tests
`test_run_flow` asserted the old call signature, so it now expects
`maxResults`. Two tests added: one where the execution is on the second page,
one where no record comes back at all.
With the source change reverted and the new tests kept, both new behaviors
fail on the bug itself:
```
FAILED test_execution_record_on_a_later_page - KeyError: 'ex_id'
FAILED test_no_matching_execution_record - KeyError: 'ex_id'
2 failed in 10.44s
```
at `hooks/appflow.py:86`, which is the `last_execs[execution_id]` lookup.
With the fix in place:
```
tests/unit/amazon/aws/hooks/test_appflow.py ..... 5 passed in 8.37s
```
`tests/unit/amazon/aws/operators/test_appflow.py` gives 6 passed and 7
errors both before and after this change. Those 7 are `db_test` cases that need
fixtures my local harness does not provide, so they are unrun rather than
passing, and they are unaffected either way. CI will cover them.
`ruff==0.16.4` (the pinned version) on both changed files: `All checks
passed!` and `2 files already formatted`.
Reproduced and tested against `apache-airflow==3.3.1`,
`apache-airflow-providers-amazon==9.35.1`, `botocore==1.43.90`, Python 3.12.14
on Linux.
### Not fixed here
The waiter acceptors filter
`flowExecutions[?executionId=='{{EXECUTION_ID}}']` out of the same call, so
with a lot of execution records the waiter can still miss the execution and
poll until `max_attempts` is exhausted on a flow that actually finished. Fixing
that properly means not using a botocore waiter for this, which felt like a
separate change. Happy to take it on if you would like it in the same PR.
closes: #72769
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
