ephraimbuddy commented on code in PR #30255:
URL: https://github.com/apache/airflow/pull/30255#discussion_r1161142078
##########
airflow/task/task_runner/base_task_runner.py:
##########
@@ -46,18 +49,23 @@ class BaseTaskRunner(LoggingMixin):
Invoke the `airflow tasks run` command with raw mode enabled in a
subprocess.
- :param local_task_job: The local task job associated with running the
- associated task instance.
+ :param base_job: The job associated with running the associated task
instance. The job_runner for it
+ should be LocalTaskJobRunner
"""
- def __init__(self, local_task_job):
- # Pass task instance context into log handlers to setup the logger.
- super().__init__(local_task_job.task_instance)
- self._task_instance = local_task_job.task_instance
+ def __init__(self, base_job: BaseJob):
+ self.job_runner: LocalTaskJobRunner = cast(LocalTaskJobRunner,
base_job.job_runner)
+ if not hasattr(self.job_runner, "task_instance"):
+ raise ValueError(
+ "BaseTaskRunner can only be used with LocalTaskJobRunner and "
+ "have task_instance field defined"
+ )
+ super().__init__(self.job_runner.task_instance)
+ self._task_instance = self.job_runner.task_instance
popen_prepend = []
if self._task_instance.run_as_user:
- self.run_as_user = self._task_instance.run_as_user
+ self.run_as_user: str | None = self._task_instance.run_as_user
Review Comment:
```suggestion
self.run_as_user: str = self._task_instance.run_as_user
```
Looks like it cannot be None?
##########
airflow/dag_processing/manager.py:
##########
@@ -576,8 +585,11 @@ def _run_parsing_loop(self):
while True:
loop_start_time = time.monotonic()
ready = multiprocessing.connection.wait(self.waitables.keys(),
timeout=poll_time)
- if self._job:
- self._job.heartbeat()
+ # we cannot (for now) define job in _job_runner nicely due to
circular references of
+ # job and job runner, so we have to use hasattr, but we might
address it in the future
Review Comment:
```suggestion
# job and job runner, so we have to use getattr, but we might
address it in the future
```
##########
airflow/task/task_runner/__init__.py:
##########
@@ -38,11 +39,11 @@
}
-def get_task_runner(local_task_job) -> BaseTaskRunner:
+def get_task_runner(local_task_job_runner: LocalTaskJobRunner) ->
BaseTaskRunner:
"""
- Get the task runner that can be used to run the given job.
+ Get the task runner that can be used to run with the give job runner.
Review Comment:
```suggestion
Get the task runner that can be used to run with the given job runner.
```
--
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]
