dabla commented on code in PR #75272:
URL: https://github.com/apache/airflow/pull/75272#discussion_r4388608032
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -310,15 +315,29 @@ def get_task_state_store(self, context: Context) ->
TaskStateStoreAccessor | Non
return context.get("task_state_store") if self.durable else None
@staticmethod
- def get_cursor(task_state_store: TaskStateStoreAccessor) ->
PaginationCursor | None:
- return cast("PaginationCursor | None",
task_state_store.get(CURSOR_KEY))
+ def get_cursor(task_state_store: TaskStateStoreAccessor) ->
_PaginationCursor | None:
+ return cast("_PaginationCursor | None",
task_state_store.get(_CURSOR_KEY))
def pull_pages(self, task_state_store: TaskStateStoreAccessor) ->
list[Any]:
- """Return the results of the pages processed so far, kept in the task
state store."""
+ """
+ Return the results of the pages processed so far, kept in the task
state store.
+
+ A page is kept wrapped in a dict, so a page whose result is ``None``
is told apart from a page
+ that is gone, for instance because it expired under ``[state_store]
default_retention_days``.
+ A missing page clears the store, so the next try starts again from the
first page instead of
+ returning results with a hole in them.
+ """
results: list[Any] = []
cursor = self.get_cursor(task_state_store)
for page in range(cursor["pages"] if cursor else 0):
- self.append_result(results=results,
result=task_state_store.get(f"{PAGE_KEY_PREFIX}{page}"))
+ kept = task_state_store.get(f"{_PAGE_KEY_PREFIX}{page}")
+ if kept is None:
+ task_state_store.clear()
Review Comment:
Good catch, fixed in 90a6db95. The pages and the cursor are now deleted by
name (`remove_pages`), from `pull_pages` on a missing page and from `clean_up`
on both the GET and the non-GET path; `msgraph_completed` stays, so a retry of
a non-GET still returns it. The fake store in the tests lost its `clear()`, and
both store tests seed a key of their own and assert it survives, so the fake
can no longer hide this.
The same commit treats a page kept unwrapped by 15.3.0rc1 like a missing
page, so a retry across the upgrade starts again from the first page instead of
failing on `kept["result"]` until the retries run out.
##########
providers/microsoft/azure/docs/changelog.rst:
##########
@@ -27,6 +27,13 @@
Changelog
---------
+.. warning::
Review Comment:
Dropped in 90a6db95, the 15.3.0 note from #75209 covers it. That also
removes the conflict with main; I'll rebase with the next push.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/powerbi.py:
##########
@@ -165,19 +166,24 @@ def get_job_status(self, external_id: JsonValue, context:
Context) -> str:
return _REFRESH_NOT_IN_HISTORY
def is_job_active(self, status: str) -> bool:
- return status == PowerBIDatasetRefreshStatus.IN_PROGRESS
+ # Without wait_for_completion a retry does not wait for a refresh in
progress either:
+ # is_job_succeeded counts it as done.
+ return self.wait_for_completion and status ==
PowerBIDatasetRefreshStatus.IN_PROGRESS
def is_job_succeeded(self, status: str) -> bool:
- return status == PowerBIDatasetRefreshStatus.COMPLETED
+ if status == PowerBIDatasetRefreshStatus.COMPLETED:
+ return True
+ # Without wait_for_completion the task only has to start the refresh.
+ return not self.wait_for_completion and status ==
PowerBIDatasetRefreshStatus.IN_PROGRESS
Review Comment:
Kept, and the comment now says so in 90a6db95: an in-progress refresh is
logged as completed and counted in `resumable_job.already_succeeded`, and why
only that path pushes the refresh id (the reconnect path goes through
`poll_until_complete` alone, without `get_job_result`).
--
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]
