kaxil commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4394303291
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -797,9 +828,47 @@ def claim_priority_files(self) -> list[DagFileInfo]:
Default implementation reads from the metadata DB; override to source
requests from an API.
"""
- if self.api_client is not None:
+ if self.api_client is None:
+ return self._claim_priority_files()
+ bundles = {bundle.name: bundle for bundle in self._dag_bundles}
+ unavailable = self._bundles_needing_refresh | {
Review Comment:
When `_find_files_in_bundle` raises in `_refresh_dag_bundles`, the bundle
has already been discarded from `_bundles_needing_refresh`, so it isn't in
`unavailable` here and its claims still count against `_parallelism`. Nothing
releases them while the listing keeps failing. With the default
`parsing_processes` of 2, two reparse requests on such a bundle block priority
parsing for all bundles. Could the API-mode listing-failure branch add the
bundle to `_bundles_needing_refresh`, with a recovery test? Related to
https://github.com/apache/airflow/pull/74794#discussion_r4389716627.
##########
airflow-core/tests/integration/dag_processing/test_api_isolation.py:
##########
@@ -0,0 +1,314 @@
+# 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.
+
+"""Run with Breeze and PostgreSQL; the processor joins an API-only internal
network."""
Review Comment:
This never runs in CI. The core integration jobs run with `--integration
<name>`, and `skip_if_not_marked_with_integration` skips anything without a
matching `pytest.mark.integration`, which this module doesn't have. The
directory also has no `__init__.py`, unlike its siblings. Since this is the
only end-to-end check that the processor works without DB network access, could
it get a marker and a matching core integration entry, or say in the docstring
that it is a manual Breeze test?
##########
airflow-core/tests/unit/dag_processing/test_requested_work_recovery.py:
##########
@@ -0,0 +1,403 @@
+# 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.
+from __future__ import annotations
+
+from pathlib import Path
+from unittest import mock
+
+import httpx2
+import pytest
+from uuid6 import uuid7
+
+from airflow._shared.timezones import timezone
+from airflow.api_fastapi.execution_api.datamodels.dag_parsing import (
+ DagBundleInventoryResponse,
+ DagBundleStateResponse,
+ DagParseResultBody,
+ ProcessorWorkItem,
+)
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import
TaskInstance
+from airflow.callbacks.callback_requests import DagCallbackRequest,
TaskCallbackRequest
+from airflow.dag_processing.api_client import DagProcessorAPIClient
+from airflow.dag_processing.bundles.base import BaseDagBundle,
DagBundleRefreshError
+from airflow.dag_processing.bundles.local import LocalDagBundle
+from airflow.dag_processing.manager import (
+ DagFileInfo,
+ DagFileProcessorManager,
+ DeferredCallback,
+ PendingDagPublication,
+ PriorityParseClaim,
+)
+from airflow.dag_processing.processor import DagFileProcessorProcess,
DagParseSource
+
+from tests_common.test_utils.config import conf_vars
+
+
+def _make_work(bundle_name, relative_fileloc="a.py", *, callback=None):
+ return ProcessorWorkItem(
+ id=str(uuid7()),
+ claim_id=uuid7(),
+ bundle_name=bundle_name,
+ relative_fileloc=relative_fileloc,
+ callback=callback.model_dump_json() if callback is not None else None,
+ )
+
+
[email protected]
+def api_manager(tmp_path):
+ ready_path = tmp_path / "ready"
+ ready_path.mkdir()
+ bundle = LocalDagBundle(name="ready", path=str(ready_path),
refresh_interval=0)
+ bundle.is_initialized = True
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ client.restart_required = False
+ client.claim_work.return_value = []
+ client.get_bundles.return_value = [
+ DagBundleStateResponse(
+ name=bundle.name, version=None, last_refreshed=None,
revision=None, team_name=None
+ )
+ ]
+ client.publish_inventory.side_effect = lambda name, body:
DagBundleInventoryResponse(
+ attempt_id=body.attempt_id, revision=uuid7(),
accepted_at=timezone.utcnow()
+ )
+ manager = DagFileProcessorManager(
+ max_runs=1, api_client=client, parallelism=2,
bundle_refresh_check_interval=0
+ )
+ manager._dag_bundles = [bundle]
+ manager._bundle_parse_sources[bundle.name] = DagParseSource()
+ return manager, client, bundle
+
+
[email protected]("bundle_version", [None, "v1"])
[email protected](DagFileProcessorProcess, "start", autospec=True)
[email protected](DagFileProcessorManager, "_get_logger_for_dag_file",
autospec=True)
[email protected](DagFileProcessorManager, "prepare_callback_bundle",
autospec=True)
+def test_callback_queued_behind_running_callback_dispatches_and_acknowledges(
+ prepare, get_logger, start, api_manager, bundle_version
+):
+ manager, client, bundle = api_manager
+ prepare.return_value = bundle
+ get_logger.return_value = (mock.Mock(spec=[]), mock.Mock(spec=[]))
+
+ def start_process(**kwargs):
+ process = mock.create_autospec(DagFileProcessorProcess, instance=True)
+ process.id = kwargs["id"]
+ process.had_callbacks = True
+ process.parse_source = kwargs["parse_source"]
+ return process
+
+ start.side_effect = start_process
+ requests = [
+ DagCallbackRequest(
+ filepath="a.py", bundle_name=bundle.name,
bundle_version=bundle_version, dag_id="dag", run_id=run
+ )
+ for run in ("first", "second")
+ ]
+ work = [_make_work(bundle.name, callback=request) for request in requests]
+ for item in work:
+ client.claim_work.return_value = [item]
+ for request in manager.fetch_callbacks():
+ manager._add_callback_to_queue(request)
+ manager._start_new_processes()
+
+ file, first = next(iter(manager._processors.items()))
+ assert start.call_count == 1
+ assert list(manager._file_queue) == [file]
+ assert [request.run_id for request in manager._callback_to_execute[file]]
== ["second"]
+ assert list(manager._callback_claims.values()) == [work[1]]
+
+ manager._processors.pop(file)
+ manager._finish_callback_attempt(first)
+ manager._start_new_processes()
+ second = manager._processors.pop(file)
+ manager._finish_callback_attempt(second)
+ manager._acknowledge_requested_work()
+
+ assert [call.kwargs["callbacks"][0].run_id for call in
start.call_args_list] == ["first", "second"]
+ assert client.acknowledge_work.call_args_list == [mock.call("callbacks",
item) for item in work]
+ assert not manager._callback_claims
+ assert not manager._callback_attempts
+ assert not manager._callback_to_execute
+ assert not manager._file_queue
+
+
[email protected](DagFileProcessorManager, "_create_process", autospec=True)
+def
test_queued_callback_preserves_db_priority_request_behind_running_parse(create,
api_manager):
+ manager, _, bundle = api_manager
+ manager.api_client = None
+ file = DagFileInfo(bundle_name=bundle.name, bundle_path=bundle.path,
rel_path=Path("a.py"))
+ running = mock.create_autospec(DagFileProcessorProcess, instance=True)
+ manager._processors[file] = running
+ manager._add_files_to_queue([file], mode="frontprio")
+ request = DagCallbackRequest(
+ filepath="a.py", bundle_name=bundle.name, bundle_version=None,
dag_id="dag", run_id="run"
+ )
+ manager._add_callback_to_queue(request)
+
+ def create_process(_, file, *, source):
+ process = mock.create_autospec(DagFileProcessorProcess, instance=True)
+ process.id = uuid7()
+ process.had_callbacks = bool(manager._callback_to_execute.pop(file,
[]))
+ process.parse_source = source
+ return process
+
+ create.side_effect = create_process
+ manager._start_new_processes()
+ create.assert_not_called()
+ assert file in manager._file_queue
+ assert file in manager._priority_files
+
+ del manager._processors[file]
+ manager._start_new_processes()
+ assert manager._processors[file].had_callbacks
+ assert file in manager._priority_files
+ assert file in manager._files_awaiting_source
+
+ del manager._processors[file]
+ manager._start_new_processes()
+ assert not manager._processors[file].had_callbacks
+ assert create.call_count == 2
+ assert file not in manager._priority_files
+ assert file not in manager._files_awaiting_source
+
+
[email protected]("failure", ["initialize", "refresh_changed",
"refresh_unchanged"])
[email protected](DagFileProcessorManager, "_create_process", autospec=True)
+def test_failed_bundle_claims_allow_healthy_priority_work_then_resume(create,
api_manager, tmp_path, failure):
+ manager, client, ready = api_manager
+ broken = mock.create_autospec(BaseDagBundle, instance=True)
+ broken.name = "broken"
+ broken.path = tmp_path / "broken"
+ broken.path.mkdir()
+ broken.is_initialized = True
+ broken.supports_versioning = False
+ broken.refresh_changes_source = True
+ broken.refresh_interval = 0
+ manager._dag_bundles.append(broken)
+ client.get_bundles.return_value.append(
+ DagBundleStateResponse(
+ name=broken.name, version=None, last_refreshed=None,
revision=None, team_name=None
+ )
+ )
+ manager._bundle_parse_sources[broken.name] = DagParseSource()
+ for filename in ("a.py", "b.py"):
+ (broken.path / filename).write_text("from airflow.sdk import DAG\ndag
= DAG('broken')\n")
+ (ready.path / "ready.py").write_text("from airflow.sdk import DAG\ndag =
DAG('ready')\n")
+ work = [_make_work(broken.name, filename) for filename in ("a.py", "b.py")]
+ client.claim_work.return_value = work
+ manager._queue_requested_files_for_parsing()
+ if failure == "initialize":
+ broken.is_initialized = False
+ broken.initialize.side_effect = OSError("unavailable")
+ else:
+ broken.refresh.side_effect = DagBundleRefreshError(
+ "unavailable", source_unchanged=failure == "refresh_unchanged"
+ )
+ known_files = {}
+ manager._refresh_dag_bundles(known_files)
+
+ assert manager._bundles_needing_refresh == {broken.name}
+ assert len(manager._priority_claims) == 2
+ assert not manager._pending_work_acks
+ healthy_work = _make_work(ready.name, "ready.py")
+ client.claim_work.return_value = [healthy_work]
+ manager._queue_requested_files_for_parsing()
+ client.claim_work.assert_called_with("priority", [ready.name], 2)
+ client.claim_work.return_value = []
+ for _ in range(3):
+ manager.claim_priority_files()
+ client.claim_work.assert_called_with("priority", [ready.name], 1)
+ assert len(manager._priority_claims) == 3
+
+ def create_process(_, file, *, source):
+ process = mock.create_autospec(DagFileProcessorProcess, instance=True)
+ process.id = uuid7()
+ process.had_callbacks = False
+ process.parse_source = source
+ return process
+
+ create.side_effect = create_process
+ manager._start_new_processes()
+ create.assert_not_called()
+ manager._refresh_dag_bundles(known_files)
+ manager._start_new_processes()
+ assert {file.bundle_name for file in manager._processors} == {ready.name}
+ _, process = manager._processors.popitem()
+ manager._finish_priority_claims(process.id)
+ manager._acknowledge_requested_work()
+ client.acknowledge_work.assert_called_once_with("priority", healthy_work)
+ assert len(manager._priority_claims) == 2
+
+ broken.initialize.side_effect = lambda: setattr(broken, "is_initialized",
True)
+ broken.refresh.side_effect = None
+ manager._refresh_dag_bundles(known_files)
+ manager._start_new_processes()
+
+ assert not manager._bundles_needing_refresh
+ assert len(manager._processors) == 2
+ assert {file.bundle_name for file in manager._processors} == {broken.name}
+ for file, process in list(manager._processors.items()):
+ claim = next(claim for claim in manager._priority_claims.values() if
claim.attempt_id == process.id)
+ assert process.parse_source.inventory_sequence >=
claim.inventory_sequence
+ manager._finish_priority_claims(process.id)
+ del manager._processors[file]
+ manager._acknowledge_requested_work()
+ assert not manager._priority_claims
+ assert client.acknowledge_work.call_count == 1 + len(work)
+ client.acknowledge_work.assert_has_calls([mock.call("priority", item) for
item in work], any_order=True)
+
+
+def
test_priority_claims_waiting_for_fresh_inventory_keep_the_budget_bounded(api_manager):
+ manager, client, bundle = api_manager
+ client.claim_work.return_value = [_make_work(bundle.name, filename) for
filename in ("a.py", "b.py")]
+ manager._queue_requested_files_for_parsing()
+
+ assert manager.claim_priority_files() == []
+
+ client.claim_work.assert_called_once_with("priority", [bundle.name], 2)
+ assert len(manager._priority_claims) == 2
+
+
[email protected]("stage", ["running", "publishing", "acknowledging",
"stale"])
+def
test_failed_bundle_only_suspends_priority_claims_without_active_attempts(api_manager,
stage):
+ manager, client, bundle = api_manager
+ manager._parallelism = 1
+ manager._bundles_needing_refresh.add(bundle.name)
+ file = DagFileInfo(bundle_name=bundle.name, bundle_path=bundle.path,
rel_path=Path("a.py"))
+ work = _make_work(bundle.name)
+ attempt_id = uuid7()
+ if stage == "acknowledging":
+ manager._pending_work_acks.append(("priority", work))
+ else:
+ manager._priority_claims[work.id, work.claim_id] = PriorityParseClaim(
+ work=work, inventory_sequence=1, attempt_id=attempt_id
+ )
+ if stage == "running":
+ process = mock.create_autospec(DagFileProcessorProcess, instance=True)
+ process.id = attempt_id
+ manager._processors[file] = process
+ elif stage == "publishing":
+ pending = mock.create_autospec(PendingDagPublication, instance=True)
+ pending.body = mock.create_autospec(DagParseResultBody, instance=True)
+ pending.body.attempt_id = attempt_id
+ manager._pending_publications[file] = pending
+
+ assert manager.claim_priority_files() == []
+
+ if stage == "stale":
+ client.claim_work.assert_called_once_with("priority", [], 1)
+ assert manager._priority_claims[work.id, work.claim_id].attempt_id ==
attempt_id
+ else:
+ client.claim_work.assert_not_called()
+
+
+def
test_uninitialized_bundle_is_not_claimed_until_initialization_recovers(api_manager):
+ manager, client, bundle = api_manager
+ bundle.is_initialized = False
+
+ manager.claim_priority_files()
+
+ client.claim_work.assert_called_once_with("priority", [], 2)
+ manager._refresh_dag_bundles({})
+ assert bundle.is_initialized
+ manager.claim_priority_files()
+ client.claim_work.assert_called_with("priority", [bundle.name], 2)
+
+
[email protected]("kind", ["callbacks", "priority"])
[email protected]("failure", ["connection", "503"])
+def test_claim_failure_preserves_deferred_callbacks_and_claims(api_manager,
kind, failure):
Review Comment:
The `priority` case seeds and asserts only callback state, which
`claim_priority_files` never touches, so it passes whatever that branch does on
a 503. Could it seed `_priority_claims` and `_priority_files` and assert they
come back unchanged?
##########
airflow-core/src/airflow/dag_processing/health.py:
##########
@@ -0,0 +1,61 @@
+# 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.
+
+"""Local probes for a processor that does not have metadata database
credentials."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from time import monotonic
+
+from airflow.configuration import conf
+
+
+def write_processor_health(last_api_heartbeat: float) -> None:
Review Comment:
This runs on every heartbeat with no `OSError` handling, so a read-only home
directory (where the default `health_check_file` lives) or a full disk crashes
a processor that is parsing and heartbeating fine. Could the write failure be
logged with the setting name and left to the probe to report?
##########
airflow-core/docs/security/security_model.rst:
##########
@@ -775,6 +786,12 @@ model — Airflow does not enforce these natively.
value itself. However, access to the file it points to should be restricted
to the Scheduler
(which generates tokens) and the API Server (which validates them).
+ A Dag File Processor running with ``[dag_processor]
execution_api_token_file`` needs none of the
+ metadata database variables, the Fernet key or JWT signing keys; it
authenticates with the token
+ file instead. It also no longer needs asset manager credentials for
publishing parsing results.
Review Comment:
The processor still builds the asset manager at import (`collection.py`
imports `airflow.assets.manager`, which calls `resolve_asset_manager()` at
module level), so a custom `asset_manager_class` still runs there and fails if
its kwargs are removed. Could this say to remove both `asset_manager_class` and
`asset_manager_kwargs` from the processor, or should the resolution become lazy?
##########
airflow-core/src/airflow/jobs/dag_processor_job_runner.py:
##########
@@ -49,12 +62,102 @@ def __init__(
):
super().__init__(job)
self.processor = processor
+ self._last_api_heartbeat = 0.0
+ self._next_api_heartbeat = 0.0
self.processor.heartbeat = lambda: perform_heartbeat(
job=self.job,
heartbeat_callback=self.heartbeat_callback,
only_if_necessary=True,
)
+ def run_with_api(self, client: DagProcessorAPIClient) -> int | None:
+ """Mirror ``run_job`` metrics and listener ordering while the API owns
Job persistence."""
+ self.processor.api_client = client
+ self.job.id = self._register_api_job(client)
+ self.job.state = JobState.RUNNING
+ self.job.start_date = self.job.latest_heartbeat = timezone.utcnow()
+ self._last_api_heartbeat = monotonic()
+ self._next_api_heartbeat = self._last_api_heartbeat +
self.job.heartrate
+ self.processor.heartbeat = lambda: self._heartbeat_api(client)
+ stats.incr("job_start", 1, 1)
+ failed = True
+ try:
+ result = execute_job(self.job, execute_callable=self._execute)
+ failed = False
+ return result
+ finally:
+ self.job.state = JobState.FAILED if failed else JobState.SUCCESS
+ try:
+ get_listener_manager().hook.before_stopping(component=self.job)
+ except Exception:
+ self.log.exception("Error calling Dag processor shutdown
listener")
+ self.job.end_date = timezone.utcnow()
+ try:
+ client.complete_job(TerminalJobState(self.job.state))
+ except Exception:
+ if not failed:
+ raise
+ self.log.exception("Unable to record the failed Dag processor
Job")
+ finally:
+ stats.incr("job_end", 1, 1)
+
+ def _register_api_job(self, client: DagProcessorAPIClient) -> int:
+ # Client imports stay local: its versioned routes import this runner.
+ from airflow.dag_processing.api_client import
DagProcessorJobAlreadyRunning
+
+ deadline = monotonic() + conf.getint("dag_processor",
"health_check_threshold") + self.job.heartrate
+ delay = 1.0
+ while True:
+ try:
+ return client.register_job()
+ except DagProcessorJobAlreadyRunning:
+ remaining = deadline - monotonic()
+ if remaining <= 0:
+ raise
+ self.log.info("Waiting for the previous Dag processor Job
heartbeat to expire")
+ sleep(min(delay, remaining))
+ delay = min(delay * 2, 5.0)
+
+ def _heartbeat_api(self, client: DagProcessorAPIClient) -> None:
+ from airflow.dag_processing.api_client import
DagProcessorRegistrationRetired
Review Comment:
This local import needs the same cycle comment as the one in
`_register_api_job`.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -856,9 +925,80 @@ def fetch_callbacks(self) -> list[CallbackRequest]:
Default implementation reads from the metadata DB; override to source
callbacks from an API.
"""
- if self.api_client is not None:
+ if self.api_client is None:
+ return self._fetch_callbacks_from_db()
+ available = (
+ self.max_callbacks_per_loop
+ - len(self._callback_claims)
+ - sum(len(work) for work in self._callback_attempts.values())
+ - sum(kind == "callbacks" for kind, _ in self._pending_work_acks)
+ )
+ if available <= 0:
+ return list(self._deferred_api_callbacks)
+ bundles = [b.name for b in self._dag_bundles if not
b.supports_versioning or b.is_initialized]
+ try:
+ work = self.api_client.claim_work("callbacks", bundles, available)
+ except httpx2.HTTPError:
+ self.log.warning("Unable to claim callbacks; retrying on a later
loop")
+ return list(self._deferred_api_callbacks)
+ for item in work:
+ if item.callback is None:
+ raise ValueError("Callback claim returned no request")
+ request: CallbackRequest =
TypeAdapter(CallbackRequest).validate_json(item.callback)
+ self._callback_claims[id(request)] = item
+ self._deferred_api_callbacks.append(request)
+ return list(self._deferred_api_callbacks)
+
+ def _acknowledge_requested_work(self, *, deadline: float | None = None) ->
None:
+ """Send queued acknowledgments in order until the deadline, keeping
claims after a failure."""
+ if self.api_client is None:
+ return
+ if deadline is None:
+ deadline = time.monotonic() + _API_SEND_BUDGET_SECONDS
+ while self._pending_work_acks and time.monotonic() < deadline:
+ kind, item = self._pending_work_acks[0]
+ try:
+ self.api_client.acknowledge_work(kind, item)
+ except httpx2.HTTPStatusError as error:
+ if error.response.status_code >= 500:
+ self.log.warning("Unable to acknowledge requested work;
retaining the claim")
+ return
+ if error.response.status_code != 409:
+ # Retrying cannot succeed; the server releases the claim
when this Job ends.
Review Comment:
A 408 or 429 from a gateway in front of the API lands here and the
acknowledgment is dropped, so the claim stays on the server until the Job ends
and the callback runs again. Could 408 and 429 be retried like a 5xx?
##########
airflow-core/src/airflow/jobs/dag_processor_job_runner.py:
##########
@@ -49,12 +62,102 @@ def __init__(
):
super().__init__(job)
self.processor = processor
+ self._last_api_heartbeat = 0.0
+ self._next_api_heartbeat = 0.0
self.processor.heartbeat = lambda: perform_heartbeat(
job=self.job,
heartbeat_callback=self.heartbeat_callback,
only_if_necessary=True,
)
+ def run_with_api(self, client: DagProcessorAPIClient) -> int | None:
+ """Mirror ``run_job`` metrics and listener ordering while the API owns
Job persistence."""
+ self.processor.api_client = client
+ self.job.id = self._register_api_job(client)
+ self.job.state = JobState.RUNNING
+ self.job.start_date = self.job.latest_heartbeat = timezone.utcnow()
+ self._last_api_heartbeat = monotonic()
+ self._next_api_heartbeat = self._last_api_heartbeat +
self.job.heartrate
+ self.processor.heartbeat = lambda: self._heartbeat_api(client)
+ stats.incr("job_start", 1, 1)
+ failed = True
+ try:
+ result = execute_job(self.job, execute_callable=self._execute)
+ failed = False
+ return result
+ finally:
+ self.job.state = JobState.FAILED if failed else JobState.SUCCESS
+ try:
+ get_listener_manager().hook.before_stopping(component=self.job)
+ except Exception:
+ self.log.exception("Error calling Dag processor shutdown
listener")
+ self.job.end_date = timezone.utcnow()
+ try:
+ client.complete_job(TerminalJobState(self.job.state))
+ except Exception:
+ if not failed:
+ raise
+ self.log.exception("Unable to record the failed Dag processor
Job")
Review Comment:
After `DagProcessorRegistrationRetired`, the API has already closed the Job,
so `complete_job` gets a 403 and this logs a full traceback on every
retire/restart. Could the `complete_job` call be skipped when
`client.restart_required` is set? The two `DagProcessorRegistrationRetired`
messages would also help with the elapsed time and `[dag_processor]
job_heartbeat_timeout`, since a long bundle refresh followed by the scheduler
ending the Job is the usual cause.
--
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]
