ephraimbuddy commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4377740510
##########
airflow-core/src/airflow/jobs/dag_processor_job_runner.py:
##########
@@ -49,12 +61,92 @@ 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:
Review Comment:
I've kept this separate and gone with the fallback. The docstring now says
it mirrors `run_job`, and shutdown follows the same order as
`complete_execution`: final state, then `before_stopping`, then `end_date`.
`test_shutdown_listener_observes_terminal_state_before_end_date` runs both
paths and checks that order.
`execute_job` (which sets the final state) and `on_starting` (fired in
`Job.__init__`) are already shared. What's left differs on purpose:
- The API only accepts a terminal state, so a `KeyboardInterrupt` is
recorded as FAILED. DB mode leaves the row RUNNING.
- If `complete_job` fails after a failed run, the original error is kept. In
`run_job`, a failing `complete_execution` replaces it.
- Registration waits for a crashed Job's heartbeat to expire. DB mode just
inserts the row.
Moving these into `run_job` would change how the scheduler and triggerer
shut down, to share about three lines. I'd revisit it if the triggerer gets an
API mode, since a second API caller would show what the interface needs.
--
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]
