amoghrajesh commented on code in PR #75272:
URL: https://github.com/apache/airflow/pull/75272#discussion_r4346122085
##########
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:
Actually, now that I think of this .... `task_state_store.clear()` deletes
every key for the task instance, not just the operator's three! A Dag that
writes its own key from `result_processor` or a subclass loses it on a durable
GET, while the same key survives a non-GET because clean_up deletes selectively
there. Delete `_CURSOR_KEY` and the page keys by name instead, here and below.
##########
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:
`is_job_succeeded` returns `True` for a refresh that is still `IN_PROGRESS`
when `wait_for_completion` is false. The mixin then logs "Job already completed
successfully, skipping resubmission" and increments
`resumable_job.already_succeeded` for a running refresh, so that metric stops
meaning what its name says. It is the cheapest way to get exactly one xcom push
on all four paths, so keep it if you prefer, but say so in the comment on line
176.
##########
providers/microsoft/azure/docs/changelog.rst:
##########
@@ -27,6 +27,13 @@
Changelog
---------
+.. warning::
Review Comment:
This new `.. warning::` goes in right under the Changelog header, which on
the current main is directly above the 15.3.0 section whose `.. note::` already
says the same thing. After rebase, the page shows both, with the warning
floating above every version header instead of belonging to a release. Drop the
changelog portion - #75209 already covers it
--
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]
