kaxil commented on code in PR #73554:
URL: https://github.com/apache/airflow/pull/73554#discussion_r4108927229
##########
airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.tsx:
##########
@@ -129,7 +136,11 @@ export const TaskInstance = () => {
return (
<ReactFlowProvider>
- <DetailsLayout error={error} isLoading={isLoading} tabs={displayTabs}>
+ <DetailsLayout
+ error={error}
+ isLoading={isLoading}
+ tabs={displayTabs.map((tab) => ({ ...tab, search: trySearch }))}
Review Comment:
This carries `try_number` into every tab, and the Audit Log and Mapped Task
Instances tabs read it as a list filter (`Events.tsx:201`,
`TaskInstances.tsx:300`). After picking a try in Logs, those tabs quietly show
only that try's rows, and the mapped list hides map indexes that are on a
different try. Could `search` go only on the tabs that use it to choose a try
(Logs, Details, Required Actions)?
##########
airflow-core/src/airflow/migrations/versions/0138_3_4_0_allocate_pending_task_attempt_numbers.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Allocate attempt numbers for pending retries and cleared tasks.
+
+Revision ID: a61f0c9d2b47
+Revises: c9f4b3e7a218
+Create Date: 2026-09-15 12:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "a61f0c9d2b47"
+down_revision = "c9f4b3e7a218"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+_TASK_INSTANCE = sa.table("task_instance", sa.column("state", sa.String),
sa.column("try_number", sa.Integer))
+
+
+def upgrade():
+ """Allocate the attempt numbers previously assigned when these tasks were
scheduled."""
+ op.execute(
+ _TASK_INSTANCE.update()
+ .where(
+ sa.or_(
+ _TASK_INSTANCE.c.state == "up_for_retry",
Review Comment:
Not every `up_for_retry` row has a new UUID yet. In 3.3.2,
`fetch_handle_failure_context` only called `prepare_db_for_next_try` when the
TI was RUNNING (`taskinstance.py:1913` at the tag), so a task the executor
killed while QUEUED or SCHEDULED went to `up_for_retry` at (A, N) with no
history row. This moves it to (A, N+1), which reuses A for the next try and
leaves try N unarchived, so `/tries/N` and its logs are gone. Could the
migration archive and rotate the `up_for_retry` rows that have no history row
at try N, or at least document the gap?
##########
airflow-core/src/airflow/ui/src/components/TaskTrySelect.tsx:
##########
@@ -78,10 +78,7 @@ export const TaskTrySelect = ({ onSelectTryNumber,
selectedTryNumber, taskInstan
(tiHistory?.task_instances ?? []).filter((ti) => ti.try_number >
0).map((ti) => [ti.try_number, ti]),
);
- if (finalTryNumber > 0 && state !== "up_for_retry" && state !== null) {
- // The current task instance is authoritative when it is also present in
history.
- triesByNumber.set(finalTryNumber, taskInstance);
- }
+ triesByNumber.set(finalTryNumber, taskInstance);
Review Comment:
This works around the public `/tries` endpoint, which still leaves out the
live row when it is `up_for_retry`
(`core_api/routes/public/task_instances.py:386-389`, "since they have been
recorded in TaskInstanceHistory"). That comment is no longer true: the live row
is now try N+1 and has no history row, so GET TI returns try N+1 and
`/tries/N+1` works, but the list omits it, while a cleared null-state N+1 row
is listed. Worth dropping that filter, and updating
`test_ti_in_retry_state_not_returned`, so API clients see the same tries as the
UI?
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1674,9 +1675,21 @@ def _on_child_started(
ti_context = self.client.task_instances.start(ti.id, self.pid,
datetime.now(tz=timezone.utc))
self._should_retry = ti_context.should_retry
self._last_successful_heartbeat = time.monotonic()
- except Exception:
+ except Exception as e:
# On any error kill that subprocess!
self.kill(signal.SIGKILL)
+ if (
+ isinstance(e, ServerResponseError)
+ and e.response.status_code == HTTPStatus.CONFLICT
+ and isinstance(e.detail, dict)
+ and e.detail.get("reason") == "invalid_state"
+ and e.detail.get("previous_state") == "restarting"
Review Comment:
This handles a clear that is noticed on the start call, but not one that
lands after the last heartbeat and before the task's own final report. Then
`succeed`/`finish`/`retry` get the same 409 with `previous_state=restarting`,
the message stays pending, and the replay in `update_task_state_if_needed`
either raises out of `wait()` (`TaskState`, `RetryTask`) or logs "TI may be
stuck on the server", and no ack is sent. The executor-event path now releases
the clear either way, so the outcome is right, but any task shorter than one
heartbeat interval takes this path with a misleading error. Could the
terminal-report calls treat this 409 like the start call and switch to the ack?
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1722,13 +1735,28 @@ def wait(self) -> int:
# Now at the last possible moment, when all logs and comms with
the subprocess has finished,
# lets upload the remote logs. Run this in a `finally` so the logs
are uploaded even if the
# state update above raised — a failed state update is exactly
when the logs matter most.
+ self._wait_completed = True
self._upload_logs()
return self._exit_code
def update_task_state_if_needed(self):
if self._terminal_state == SERVER_TERMINATED:
self._pending_terminal_state_msg = None
+ try:
+ self.client.task_instances.finish(
+ id=self.id,
+ state=SERVER_TERMINATED,
+ when=datetime.now(tz=timezone.utc),
+ rendered_map_index=self._rendered_map_index,
+ pid=self.pid,
+ )
+ except ServerResponseError as error:
+ if error.response.status_code != HTTPStatus.NOT_FOUND:
Review Comment:
Thanks for the 404 handling. A 409 still raises here, though: when the
heartbeat got 409 `running_elsewhere`, this ack runs the same hostname/pid
check at `routes/task_instances.py:415`, so it is certain to get 409 and
`ServerResponseError` comes out of `wait()`. On main that path just returned
the exit code. Should a 409 on the ack also mean "nothing to acknowledge"?
--
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]
